test(wave115): 抽离watermark_config领域模型 + 76单测

- 抽离WatermarkConfig数据类、9宫格位置常量、位置计算、滤镜构建到packages/domain/watermark_config.py
- 新增has_effect()/get_position_names()/get_position_display_name()工具函数
- watermark_engine.py: 314→117行 (-63%)，保留WatermarkEngine薄包装+模块级向后兼容导入
- 76个纯逻辑单测全绿，原有48个测试无回归
- 合计124 passed

apps/worker/video_processing/watermark_engine.py

@@ -6,135 +6,33 @@
 - 9宫格位置 + 边距配置
 - 透明度/大小缩放
 - 滚动水印（跑马灯）
+
+注：核心领域模型已抽离到 packages/domain/watermark_config.py，
+本模块保留薄包装层，确保向后兼容。
 """
 
 from __future__ import annotations
 
 import logging
-from dataclasses import dataclass
 from typing import Any
 
+from packages.domain.watermark_config import (  # noqa: F401 — 向后兼容
+    WATERMARK_POSITIONS,
+    WatermarkConfig,
+    build_image_watermark_filter as _build_image_watermark_filter,
+    build_text_watermark_filter as _build_text_watermark_filter,
+    calc_position as _calc_position_base,
+    calc_scroll_x as _calc_scroll_x_base,
+)
+
 logger = logging.getLogger(__name__)
 
-# 9宫格位置枚举
-WATERMARK_POSITIONS = {
-    "top_left": "左上",
-    "top_center": "中上",
-    "top_right": "右上",
-    "center_left": "左中",
-    "center": "中心",
-    "center_right": "右中",
-    "bottom_left": "左下",
-    "bottom_center": "中下",
-    "bottom_right": "右下",
-}
-
-
-@dataclass
-class WatermarkConfig:
-    """水印配置.
-
-    mode: "image" 图片水印 | "text" 文字水印
-    position: 9宫格位置
-    opacity: 透明度 0.0-1.0
-    scale: 缩放比例（图片水印），0.1-1.0
-    margin: 边距（像素）
-    scroll: 是否滚动（跑马灯）
-    scroll_speed: 滚动速度（像素/秒）
-    """
-
-    mode: str = "text"  # image | text
-    position: str = "bottom_right"
-
-    # 图片水印
-    image_path: str = ""  # 本地图片路径
-    scale: float = 0.2  # 相对输出宽度的比例
-    opacity: float = 0.8  # 0.0-1.0
-
-    # 文字水印
-    text: str = ""
-    font_size: int = 24
-    font_color: str = "white"
-    font_path: str = ""  # 字体文件路径
-
-    # 边距
-    margin_x: int = 20
-    margin_y: int = 20
-
-    # 滚动水印
-    scroll: bool = False
-    scroll_speed: int = 50  # 像素/秒
-
-    @classmethod
-    def from_dict(cls, data: dict[str, Any] | None) -> WatermarkConfig | None:
-        """从字典构造，空配置返回 None（不加水印）."""
-        if not data:
-            return None
-
-        enabled = data.get("enabled", False)
-        if not enabled:
-            return None
-
-        mode = data.get("mode", "text")
-
-        # 图片模式需要 image_path；文字模式需要 text
-        if mode == "image":
-            image_path = data.get("image_path", "") or data.get("image", "") or ""
-            if not image_path:
-                logger.warning("图片水印缺少 image_path，跳过水印")
-                return None
-        elif mode == "text":
-            text = data.get("text", "") or ""
-            if not text:
-                logger.warning("文字水印缺少 text，跳过水印")
-                return None
-
-        position = data.get("position", "bottom_right")
-        if position not in WATERMARK_POSITIONS:
-            position = "bottom_right"
-
-        return cls(
-            mode=mode,
-            position=position,
-            image_path=str(data.get("image_path", data.get("image", "")) or ""),
-            scale=float(data.get("scale", 0.2)),
-            opacity=float(data.get("opacity", 0.8)),
-            text=str(data.get("text", "") or ""),
-            font_size=int(data.get("font_size", 24)),
-            font_color=str(data.get("font_color", "white")),
-            font_path=str(data.get("font_path", "") or ""),
-            margin_x=int(data.get("margin_x", 20)),
-            margin_y=int(data.get("margin_y", 20)),
-            scroll=bool(data.get("scroll", False)),
-            scroll_speed=int(data.get("scroll_speed", 50)),
-        )
-
-    def validate(self) -> tuple[bool, str]:
-        """校验配置是否有效."""
-        if self.position not in WATERMARK_POSITIONS:
-            return False, f"不支持的位置: {self.position}"
-
-        if not (0.0 <= self.opacity <= 1.0):
-            return False, "透明度必须在 0-1 之间"
-
-        if self.mode == "image":
-            if not self.image_path:
-                return False, "图片水印缺少图片路径"
-            if not (0.01 <= self.scale <= 1.0):
-                return False, "缩放比例必须在 0.01-1.0 之间"
-        elif self.mode == "text":
-            if not self.text:
-                return False, "文字水印缺少文字内容"
-            if self.font_size <= 0:
-                return False, "字体大小必须大于 0"
-        else:
-            return False, f"不支持的水印模式: {self.mode}"
-
-        return True, ""
-
 
 class WatermarkEngine:
-    """水印引擎 — 生成 FFmpeg 水印滤镜."""
+    """水印引擎 — 生成 FFmpeg 水印滤镜.
+
+    薄包装层，实际逻辑委托给 packages.domain.watermark_config。
+    """
 
     @staticmethod
     def calc_position(
@@ -150,27 +48,7 @@ class WatermarkEngine:
 
         坐标系：左上角为 (0, 0)
         """
-        if position == "top_left":
-            return margin_x, margin_y
-        elif position == "top_center":
-            return (output_width - wm_width) // 2, margin_y
-        elif position == "top_right":
-            return output_width - wm_width - margin_x, margin_y
-        elif position == "center_left":
-            return margin_x, (output_height - wm_height) // 2
-        elif position == "center":
-            return (output_width - wm_width) // 2, (output_height - wm_height) // 2
-        elif position == "center_right":
-            return output_width - wm_width - margin_x, (output_height - wm_height) // 2
-        elif position == "bottom_left":
-            return margin_x, output_height - wm_height - margin_y
-        elif position == "bottom_center":
-            return (output_width - wm_width) // 2, output_height - wm_height - margin_y
-        elif position == "bottom_right":
-            return output_width - wm_width - margin_x, output_height - wm_height - margin_y
-        else:
-            # 默认右下角
-            return output_width - wm_width - margin_x, output_height - wm_height - margin_y
+        return _calc_position_base(position, output_width, output_height, wm_width, wm_height, margin_x, margin_y)
 
     @staticmethod
     def calc_scroll_x(position: str, output_width: int, wm_width: int, speed: int) -> str:
@@ -178,12 +56,7 @@ class WatermarkEngine:
 
         从右向左滚动（跑马灯效果）
         """
-        # x 从 W 到 -wm_width，整个宽度 + wm_width 的距离
-        # 使用 overlay 的 enable 表达式
-        # x = 'W - (t * speed)' → 不对，应该是持续滚动
-        # 标准跑马灯：x = -w + (t * speed) % (W + w)
-        # 但 FFmpeg overlay 支持表达式
-        return f"mod({output_width}-mod({speed}*t\\,{output_width}+{wm_width})"
+        return _calc_scroll_x_base(position, output_width, wm_width, speed)
 
     @staticmethod
     def build_image_watermark_filter(
@@ -208,51 +81,15 @@ class WatermarkEngine:
             (filter_complex_str, input_args_list)
             input_args 是 ["-i", wm_image_path] 格式
         """
-        # 计算水印尺寸（按输出宽度比例缩放）
-        wm_width = int(output_width * config.scale)
-        wm_height = -1  # 保持比例
-        wm_filter = f"scale={wm_width}:{wm_height}"
-
-        # 透明度处理
-        if config.opacity < 1.0:
-            wm_filter += f",format=rgba,colorchannelmixer=aa={config.opacity}"
-
-        # 水印预处理标签
-        wm_pre_label = "[wm_scaled]"
-
-        # 计算位置
-        x, y = WatermarkEngine.calc_position(
-            config.position,
+        return _build_image_watermark_filter(
+            input_video_label,
+            wm_image_path,
             output_width,
             output_height,
-            wm_width,
-            wm_width,  # 高度未知，先用宽度估算
-            config.margin_x,
-            config.margin_y,
+            output_label,
+            config,
         )
 
-        # 滚动水印
-        if config.scroll:
-            # 从右向左滚动：x = W - (t * speed) mod (W + wm_w)
-            # 使用 overlay 表达式
-            x_expr = f"{output_width}-mod({config.scroll_speed}*t\\,{output_width}+{wm_width}"
-            y_expr = str(y)
-            overlay_expr = f"x={x_expr}:y={y_expr}"
-        else:
-            overlay_expr = f"x={x}:y={y}"
-
-        # 构建滤镜
-        # 先缩放水印图
-        filter_parts = [
-            f"[1:v]{wm_filter}{wm_pre_label}",
-            f"{input_video_label}{wm_pre_label}overlay={overlay_expr}{output_label}",
-        ]
-
-        filter_complex = ";".join(filter_parts)
-        input_args = ["-i", wm_image_path]
-
-        return filter_complex, input_args
-
     @staticmethod
     def build_text_watermark_filter(
         input_video_label: str,
@@ -273,42 +110,4 @@ class WatermarkEngine:
         Returns:
             FFmpeg filter 字符串
         """
-        # 转义文字中的特殊字符
-        text = config.text.replace(":", "\\:").replace("'", "\\'")
-
-        # 字体配置
-        font_config = []
-        if config.font_path:
-            font_path_escaped = config.font_path.replace(":", "\\:").replace("'", "\\'")
-            font_config.append(f"fontfile='{font_path_escaped}'")
-        font_config.append(f"fontsize={config.font_size}")
-        font_config.append(f"fontcolor={config.font_color}@{config.opacity}")
-
-        # 估算文字宽高（粗略估算，用于位置计算）
-        # 每个汉字约等于 font_size 宽高
-        approx_w = len(config.text) * config.font_size
-        approx_h = config.font_size
-
-        # 位置计算
-        x, y = WatermarkEngine.calc_position(
-            config.position,
-            output_width,
-            output_height,
-            approx_w,
-            approx_h,
-            config.margin_x,
-            config.margin_y,
-        )
-
-        # 滚动水印
-        if config.scroll:
-            x_expr = f"w-mod({config.scroll_speed}*t\\,W+w)"
-            pos_config = [f"x={x_expr}", f"y={y}"]
-        else:
-            pos_config = [f"x={x}", f"y={y}"]
-
-        # 组装 drawtext
-        drawtext_parts = [f"text='{text}'"] + font_config + pos_config
-        drawtext = "drawtext=" + ":".join(drawtext_parts)
-
-        return f"{input_video_label}{drawtext}{output_label}"
+        return _build_text_watermark_filter(input_video_label, output_label, config, output_width, output_height)

packages/domain/watermark_config.py

@@ -0,0 +1,360 @@
+"""水印配置领域模型 — 纯逻辑，无FFmpeg依赖.
+
+抽离自 watermark_engine.py，包含：
+- 水印位置常量（9宫格）
+- WatermarkConfig 数据类（from_dict / validate）
+- 位置计算（calc_position / calc_scroll_x）
+- 滤镜字符串构建（build_image_watermark_filter / build_text_watermark_filter）
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+# ── 常量 ────────────────────────────────────────────────────────────────────
+
+# 9宫格位置枚举
+WATERMARK_POSITIONS: dict[str, str] = {
+    "top_left": "左上",
+    "top_center": "中上",
+    "top_right": "右上",
+    "center_left": "左中",
+    "center": "中心",
+    "center_right": "右中",
+    "bottom_left": "左下",
+    "bottom_center": "中下",
+    "bottom_right": "右下",
+}
+
+VALID_POSITIONS = set(WATERMARK_POSITIONS.keys())
+
+# 默认值常量
+DEFAULT_POSITION = "bottom_right"
+DEFAULT_MODE = "text"
+DEFAULT_SCALE = 0.2
+DEFAULT_OPACITY = 0.8
+DEFAULT_FONT_SIZE = 24
+DEFAULT_FONT_COLOR = "white"
+DEFAULT_MARGIN_X = 20
+DEFAULT_MARGIN_Y = 20
+DEFAULT_SCROLL_SPEED = 50
+
+
+# ── 数据类 ──────────────────────────────────────────────────────────────────
+
+
+@dataclass
+class WatermarkConfig:
+    """水印配置.
+
+    mode: "image" 图片水印 | "text" 文字水印
+    position: 9宫格位置
+    opacity: 透明度 0.0-1.0
+    scale: 缩放比例（图片水印），0.1-1.0
+    margin: 边距（像素）
+    scroll: 是否滚动（跑马灯）
+    scroll_speed: 滚动速度（像素/秒）
+    """
+
+    mode: str = DEFAULT_MODE  # image | text
+    position: str = DEFAULT_POSITION
+
+    # 图片水印
+    image_path: str = ""  # 本地图片路径
+    scale: float = DEFAULT_SCALE  # 相对输出宽度的比例
+    opacity: float = DEFAULT_OPACITY  # 0.0-1.0
+
+    # 文字水印
+    text: str = ""
+    font_size: int = DEFAULT_FONT_SIZE
+    font_color: str = DEFAULT_FONT_COLOR
+    font_path: str = ""  # 字体文件路径
+
+    # 边距
+    margin_x: int = DEFAULT_MARGIN_X
+    margin_y: int = DEFAULT_MARGIN_Y
+
+    # 滚动水印
+    scroll: bool = False
+    scroll_speed: int = DEFAULT_SCROLL_SPEED  # 像素/秒
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any] | None) -> WatermarkConfig | None:
+        """从字典构造，空配置返回 None（不加水印）."""
+        if not data:
+            return None
+
+        enabled = data.get("enabled", False)
+        if not enabled:
+            return None
+
+        mode = data.get("mode", DEFAULT_MODE)
+
+        # 图片模式需要 image_path；文字模式需要 text
+        if mode == "image":
+            image_path = data.get("image_path", "") or data.get("image", "") or ""
+            if not image_path:
+                logger.warning("图片水印缺少 image_path，跳过水印")
+                return None
+        elif mode == "text":
+            text = data.get("text", "") or ""
+            if not text:
+                logger.warning("文字水印缺少 text，跳过水印")
+                return None
+
+        position = data.get("position", DEFAULT_POSITION)
+        if position not in VALID_POSITIONS:
+            position = DEFAULT_POSITION
+
+        return cls(
+            mode=mode,
+            position=position,
+            image_path=str(data.get("image_path", data.get("image", "")) or ""),
+            scale=float(data.get("scale", DEFAULT_SCALE)),
+            opacity=float(data.get("opacity", DEFAULT_OPACITY)),
+            text=str(data.get("text", "") or ""),
+            font_size=int(data.get("font_size", DEFAULT_FONT_SIZE)),
+            font_color=str(data.get("font_color", DEFAULT_FONT_COLOR)),
+            font_path=str(data.get("font_path", "") or ""),
+            margin_x=int(data.get("margin_x", DEFAULT_MARGIN_X)),
+            margin_y=int(data.get("margin_y", DEFAULT_MARGIN_Y)),
+            scroll=bool(data.get("scroll", False)),
+            scroll_speed=int(data.get("scroll_speed", DEFAULT_SCROLL_SPEED)),
+        )
+
+    def validate(self) -> tuple[bool, str]:
+        """校验配置是否有效."""
+        if self.position not in VALID_POSITIONS:
+            return False, f"不支持的位置: {self.position}"
+
+        if not (0.0 <= self.opacity <= 1.0):
+            return False, "透明度必须在 0-1 之间"
+
+        if self.mode == "image":
+            if not self.image_path:
+                return False, "图片水印缺少图片路径"
+            if not (0.01 <= self.scale <= 1.0):
+                return False, "缩放比例必须在 0.01-1.0 之间"
+        elif self.mode == "text":
+            if not self.text:
+                return False, "文字水印缺少文字内容"
+            if self.font_size <= 0:
+                return False, "字体大小必须大于 0"
+        else:
+            return False, f"不支持的水印模式: {self.mode}"
+
+        return True, ""
+
+    def has_effect(self) -> bool:
+        """判断水印是否有实际效果（非空配置）."""
+        if self.mode == "image":
+            return bool(self.image_path) and self.opacity > 0
+        elif self.mode == "text":
+            return bool(self.text) and self.opacity > 0 and self.font_size > 0
+        return False
+
+
+# ── 位置计算 ────────────────────────────────────────────────────────────────
+
+
+def calc_position(
+    position: str,
+    output_width: int,
+    output_height: int,
+    wm_width: int,
+    wm_height: int,
+    margin_x: int,
+    margin_y: int,
+) -> tuple[int, int]:
+    """根据9宫格位置计算水印坐标 (x, y).
+
+    坐标系：左上角为 (0, 0)
+    """
+    if position == "top_left":
+        return margin_x, margin_y
+    elif position == "top_center":
+        return (output_width - wm_width) // 2, margin_y
+    elif position == "top_right":
+        return output_width - wm_width - margin_x, margin_y
+    elif position == "center_left":
+        return margin_x, (output_height - wm_height) // 2
+    elif position == "center":
+        return (output_width - wm_width) // 2, (output_height - wm_height) // 2
+    elif position == "center_right":
+        return output_width - wm_width - margin_x, (output_height - wm_height) // 2
+    elif position == "bottom_left":
+        return margin_x, output_height - wm_height - margin_y
+    elif position == "bottom_center":
+        return (output_width - wm_width) // 2, output_height - wm_height - margin_y
+    elif position == "bottom_right":
+        return output_width - wm_width - margin_x, output_height - wm_height - margin_y
+    else:
+        # 默认右下角
+        return output_width - wm_width - margin_x, output_height - wm_height - margin_y
+
+
+def calc_scroll_x(position: str, output_width: int, wm_width: int, speed: int) -> str:
+    """生成滚动水印的 x 坐标表达式.
+
+    从右向左滚动（跑马灯效果）
+    """
+    # 标准跑马灯：x = -w + (t * speed) % (W + w)
+    # FFmpeg overlay 表达式写法
+    return f"mod({output_width}-mod({speed}*t\\,{output_width}+{wm_width})"
+
+
+# ── 滤镜构建 ────────────────────────────────────────────────────────────────
+
+
+def build_image_watermark_filter(
+    input_video_label: str,
+    wm_image_path: str,
+    output_width: int,
+    output_height: int,
+    output_label: str,
+    config: WatermarkConfig,
+) -> tuple[str, list[str]]:
+    """构建图片水印滤镜链.
+
+    Args:
+        input_video_label: 输入视频标签，如 "[final_video]"
+        wm_image_path: 水印图片本地路径
+        output_width: 输出视频宽度
+        output_height: 输出视频高度
+        output_label: 输出标签
+        config: 水印配置
+
+    Returns:
+        (filter_complex_str, input_args_list)
+        input_args 是 ["-i", wm_image_path] 格式
+    """
+    # 计算水印尺寸（按输出宽度比例缩放）
+    wm_width = int(output_width * config.scale)
+    wm_height = -1  # 保持比例
+    wm_filter = f"scale={wm_width}:{wm_height}"
+
+    # 透明度处理
+    if config.opacity < 1.0:
+        wm_filter += f",format=rgba,colorchannelmixer=aa={config.opacity}"
+
+    # 水印预处理标签
+    wm_pre_label = "[wm_scaled]"
+
+    # 计算位置
+    x, y = calc_position(
+        config.position,
+        output_width,
+        output_height,
+        wm_width,
+        wm_width,  # 高度未知，先用宽度估算
+        config.margin_x,
+        config.margin_y,
+    )
+
+    # 滚动水印
+    if config.scroll:
+        # 从右向左滚动：x = W - (t * speed) mod (W + wm_w)
+        x_expr = f"{output_width}-mod({config.scroll_speed}*t\\,{output_width}+{wm_width}"
+        y_expr = str(y)
+        overlay_expr = f"x={x_expr}:y={y_expr}"
+    else:
+        overlay_expr = f"x={x}:y={y}"
+
+    # 构建滤镜
+    filter_parts = [
+        f"[1:v]{wm_filter}{wm_pre_label}",
+        f"{input_video_label}{wm_pre_label}overlay={overlay_expr}{output_label}",
+    ]
+
+    filter_complex = ";".join(filter_parts)
+    input_args = ["-i", wm_image_path]
+
+    return filter_complex, input_args
+
+
+def build_text_watermark_filter(
+    input_video_label: str,
+    output_label: str,
+    config: WatermarkConfig,
+    output_width: int,
+    output_height: int,
+) -> str:
+    """构建文字水印滤镜（drawtext）.
+
+    Args:
+        input_video_label: 输入视频标签
+        output_label: 输出标签
+        config: 水印配置
+        output_width: 输出宽度
+        output_height: 输出高度
+
+    Returns:
+        FFmpeg filter 字符串
+    """
+    # 转义文字中的特殊字符
+    text = config.text.replace(":", "\\:").replace("'", "\\'")
+
+    # 字体配置
+    font_config = []
+    if config.font_path:
+        font_path_escaped = config.font_path.replace(":", "\\:").replace("'", "\\'")
+        font_config.append(f"fontfile='{font_path_escaped}'")
+    font_config.append(f"fontsize={config.font_size}")
+    font_config.append(f"fontcolor={config.font_color}@{config.opacity}")
+
+    # 估算文字宽高（粗略估算，用于位置计算）
+    # 每个汉字约等于 font_size 宽高
+    approx_w = len(config.text) * config.font_size
+    approx_h = config.font_size
+
+    # 位置计算
+    x, y = calc_position(
+        config.position,
+        output_width,
+        output_height,
+        approx_w,
+        approx_h,
+        config.margin_x,
+        config.margin_y,
+    )
+
+    # 滚动水印
+    if config.scroll:
+        x_expr = f"w-mod({config.scroll_speed}*t\\,W+w)"
+        pos_config = [f"x={x_expr}", f"y={y}"]
+    else:
+        pos_config = [f"x={x}", f"y={y}"]
+
+    # 组装 drawtext
+    drawtext_parts = [f"text='{text}'"] + font_config + pos_config
+    drawtext = "drawtext=" + ":".join(drawtext_parts)
+
+    return f"{input_video_label}{drawtext}{output_label}"
+
+
+# ── 工具函数 ────────────────────────────────────────────────────────────────
+
+
+def get_position_names() -> list[str]:
+    """获取所有合法位置名称列表（按从上到下、从左到右顺序）."""
+    return [
+        "top_left",
+        "top_center",
+        "top_right",
+        "center_left",
+        "center",
+        "center_right",
+        "bottom_left",
+        "bottom_center",
+        "bottom_right",
+    ]
+
+
+def get_position_display_name(position: str) -> str:
+    """获取位置的中文显示名."""
+    return WATERMARK_POSITIONS.get(position, position)

tests/unit/test_watermark_config.py

@@ -0,0 +1,452 @@
+"""watermark_config 领域模型单测."""
+
+from __future__ import annotations
+
+import pytest
+
+from packages.domain.watermark_config import (
+    DEFAULT_FONT_COLOR,
+    DEFAULT_FONT_SIZE,
+    DEFAULT_MODE,
+    DEFAULT_OPACITY,
+    DEFAULT_POSITION,
+    DEFAULT_SCALE,
+    VALID_POSITIONS,
+    WATERMARK_POSITIONS,
+    WatermarkConfig,
+    build_image_watermark_filter,
+    build_text_watermark_filter,
+    calc_position,
+    calc_scroll_x,
+    get_position_display_name,
+    get_position_names,
+)
+
+# ── 常量测试 ────────────────────────────────────────────────────────────────
+
+
+class TestConstants:
+    def test_nine_positions(self):
+        assert len(WATERMARK_POSITIONS) == 9
+
+    def test_all_position_keys_valid(self):
+        for key in WATERMARK_POSITIONS:
+            assert key in VALID_POSITIONS
+
+    def test_valid_positions_match(self):
+        assert set(WATERMARK_POSITIONS.keys()) == VALID_POSITIONS
+
+    def test_default_values(self):
+        assert DEFAULT_POSITION == "bottom_right"
+        assert DEFAULT_MODE == "text"
+        assert DEFAULT_SCALE == 0.2
+        assert DEFAULT_OPACITY == 0.8
+        assert DEFAULT_FONT_SIZE == 24
+        assert DEFAULT_FONT_COLOR == "white"
+
+
+# ── WatermarkConfig.from_dict 测试 ─────────────────────────────────────────
+
+
+class TestWatermarkConfigFromDict:
+    def test_none_returns_none(self):
+        assert WatermarkConfig.from_dict(None) is None
+
+    def test_empty_dict_returns_none(self):
+        assert WatermarkConfig.from_dict({}) is None
+
+    def test_disabled_returns_none(self):
+        assert WatermarkConfig.from_dict({"enabled": False}) is None
+
+    def test_text_mode_basic(self):
+        cfg = WatermarkConfig.from_dict({"enabled": True, "mode": "text", "text": "hello"})
+        assert cfg is not None
+        assert cfg.mode == "text"
+        assert cfg.text == "hello"
+        assert cfg.position == DEFAULT_POSITION
+
+    def test_image_mode_basic(self):
+        cfg = WatermarkConfig.from_dict({"enabled": True, "mode": "image", "image_path": "/tmp/wm.png"})
+        assert cfg is not None
+        assert cfg.mode == "image"
+        assert cfg.image_path == "/tmp/wm.png"
+
+    def test_image_mode_accepts_image_key(self):
+        cfg = WatermarkConfig.from_dict({"enabled": True, "mode": "image", "image": "/tmp/wm.png"})
+        assert cfg is not None
+        assert cfg.image_path == "/tmp/wm.png"
+
+    def test_image_mode_missing_path_returns_none(self):
+        assert WatermarkConfig.from_dict({"enabled": True, "mode": "image"}) is None
+
+    def test_text_mode_missing_text_returns_none(self):
+        assert WatermarkConfig.from_dict({"enabled": True, "mode": "text"}) is None
+
+    def test_text_mode_empty_text_returns_none(self):
+        assert WatermarkConfig.from_dict({"enabled": True, "mode": "text", "text": ""}) is None
+
+    def test_invalid_position_defaults(self):
+        cfg = WatermarkConfig.from_dict({"enabled": True, "mode": "text", "text": "hi", "position": "invalid"})
+        assert cfg.position == DEFAULT_POSITION
+
+    def test_custom_all_params(self):
+        cfg = WatermarkConfig.from_dict(
+            {
+                "enabled": True,
+                "mode": "text",
+                "text": "测试水印",
+                "position": "top_left",
+                "font_size": 32,
+                "font_color": "red",
+                "opacity": 0.5,
+                "margin_x": 30,
+                "margin_y": 40,
+                "scroll": True,
+                "scroll_speed": 100,
+            }
+        )
+        assert cfg is not None
+        assert cfg.text == "测试水印"
+        assert cfg.position == "top_left"
+        assert cfg.font_size == 32
+        assert cfg.font_color == "red"
+        assert cfg.opacity == 0.5
+        assert cfg.margin_x == 30
+        assert cfg.margin_y == 40
+        assert cfg.scroll is True
+        assert cfg.scroll_speed == 100
+
+    def test_default_mode_is_text(self):
+        cfg = WatermarkConfig.from_dict({"enabled": True, "text": "hi"})
+        assert cfg is not None
+        assert cfg.mode == "text"
+
+
+# ── WatermarkConfig.validate 测试 ──────────────────────────────────────────
+
+
+class TestWatermarkConfigValidate:
+    def test_valid_text_config(self):
+        cfg = WatermarkConfig(mode="text", text="hello")
+        ok, msg = cfg.validate()
+        assert ok is True
+        assert msg == ""
+
+    def test_valid_image_config(self):
+        cfg = WatermarkConfig(mode="image", image_path="/tmp/wm.png", scale=0.3)
+        ok, msg = cfg.validate()
+        assert ok is True
+
+    def test_invalid_position(self):
+        cfg = WatermarkConfig(mode="text", text="hi", position="nowhere")
+        ok, msg = cfg.validate()
+        assert ok is False
+        assert "位置" in msg
+
+    def test_opacity_negative(self):
+        cfg = WatermarkConfig(mode="text", text="hi", opacity=-0.1)
+        ok, msg = cfg.validate()
+        assert ok is False
+        assert "透明度" in msg
+
+    def test_opacity_over_one(self):
+        cfg = WatermarkConfig(mode="text", text="hi", opacity=1.5)
+        ok, msg = cfg.validate()
+        assert ok is False
+
+    def test_opacity_boundary_zero(self):
+        cfg = WatermarkConfig(mode="text", text="hi", opacity=0.0)
+        ok, _ = cfg.validate()
+        assert ok is True
+
+    def test_opacity_boundary_one(self):
+        cfg = WatermarkConfig(mode="text", text="hi", opacity=1.0)
+        ok, _ = cfg.validate()
+        assert ok is True
+
+    def test_image_missing_path(self):
+        cfg = WatermarkConfig(mode="image")
+        ok, msg = cfg.validate()
+        assert ok is False
+        assert "图片路径" in msg
+
+    def test_image_scale_too_small(self):
+        cfg = WatermarkConfig(mode="image", image_path="/a.png", scale=0.001)
+        ok, msg = cfg.validate()
+        assert ok is False
+        assert "缩放比例" in msg
+
+    def test_image_scale_too_large(self):
+        cfg = WatermarkConfig(mode="image", image_path="/a.png", scale=2.0)
+        ok, msg = cfg.validate()
+        assert ok is False
+
+    def test_image_scale_boundary_low(self):
+        cfg = WatermarkConfig(mode="image", image_path="/a.png", scale=0.01)
+        ok, _ = cfg.validate()
+        assert ok is True
+
+    def test_image_scale_boundary_high(self):
+        cfg = WatermarkConfig(mode="image", image_path="/a.png", scale=1.0)
+        ok, _ = cfg.validate()
+        assert ok is True
+
+    def test_text_missing_content(self):
+        cfg = WatermarkConfig(mode="text", text="")
+        ok, msg = cfg.validate()
+        assert ok is False
+        assert "文字内容" in msg
+
+    def test_text_font_size_zero(self):
+        cfg = WatermarkConfig(mode="text", text="hi", font_size=0)
+        ok, msg = cfg.validate()
+        assert ok is False
+        assert "字体大小" in msg
+
+    def test_text_font_size_negative(self):
+        cfg = WatermarkConfig(mode="text", text="hi", font_size=-5)
+        ok, msg = cfg.validate()
+        assert ok is False
+
+    def test_unknown_mode(self):
+        cfg = WatermarkConfig(mode="video")
+        ok, msg = cfg.validate()
+        assert ok is False
+        assert "模式" in msg
+
+
+# ── has_effect 测试 ────────────────────────────────────────────────────────
+
+
+class TestHasEffect:
+    def test_text_with_content_has_effect(self):
+        cfg = WatermarkConfig(mode="text", text="hello")
+        assert cfg.has_effect() is True
+
+    def test_text_empty_no_effect(self):
+        cfg = WatermarkConfig(mode="text", text="")
+        assert cfg.has_effect() is False
+
+    def test_text_zero_opacity_no_effect(self):
+        cfg = WatermarkConfig(mode="text", text="hello", opacity=0.0)
+        assert cfg.has_effect() is False
+
+    def test_image_with_path_has_effect(self):
+        cfg = WatermarkConfig(mode="image", image_path="/a.png")
+        assert cfg.has_effect() is True
+
+    def test_image_no_path_no_effect(self):
+        cfg = WatermarkConfig(mode="image")
+        assert cfg.has_effect() is False
+
+    def test_unknown_mode_no_effect(self):
+        cfg = WatermarkConfig(mode="unknown")
+        assert cfg.has_effect() is False
+
+
+# ── calc_position 测试 ─────────────────────────────────────────────────────
+
+
+class TestCalcPosition:
+    def test_top_left(self):
+        x, y = calc_position("top_left", 1000, 2000, 100, 50, 10, 20)
+        assert (x, y) == (10, 20)
+
+    def test_top_center(self):
+        x, y = calc_position("top_center", 1000, 2000, 100, 50, 10, 20)
+        assert (x, y) == (450, 20)
+
+    def test_top_right(self):
+        x, y = calc_position("top_right", 1000, 2000, 100, 50, 10, 20)
+        assert (x, y) == (890, 20)
+
+    def test_center_left(self):
+        x, y = calc_position("center_left", 1000, 2000, 100, 50, 10, 20)
+        assert (x, y) == (10, 975)
+
+    def test_center(self):
+        x, y = calc_position("center", 1000, 2000, 100, 50, 10, 20)
+        assert (x, y) == (450, 975)
+
+    def test_center_right(self):
+        x, y = calc_position("center_right", 1000, 2000, 100, 50, 10, 20)
+        assert (x, y) == (890, 975)
+
+    def test_bottom_left(self):
+        x, y = calc_position("bottom_left", 1000, 2000, 100, 50, 10, 20)
+        assert (x, y) == (10, 1930)
+
+    def test_bottom_center(self):
+        x, y = calc_position("bottom_center", 1000, 2000, 100, 50, 10, 20)
+        assert (x, y) == (450, 1930)
+
+    def test_bottom_right(self):
+        x, y = calc_position("bottom_right", 1000, 2000, 100, 50, 10, 20)
+        assert (x, y) == (890, 1930)
+
+    def test_unknown_position_defaults_bottom_right(self):
+        x, y = calc_position("invalid", 1000, 2000, 100, 50, 10, 20)
+        assert (x, y) == (890, 1930)
+
+    def test_zero_margin(self):
+        x, y = calc_position("top_left", 1000, 2000, 100, 50, 0, 0)
+        assert (x, y) == (0, 0)
+
+    def test_small_output(self):
+        x, y = calc_position("center", 100, 100, 50, 30, 5, 5)
+        assert (x, y) == (25, 35)
+
+
+# ── calc_scroll_x 测试 ─────────────────────────────────────────────────────
+
+
+class TestCalcScrollX:
+    def test_returns_string_expression(self):
+        result = calc_scroll_x("bottom", 1000, 200, 50)
+        assert isinstance(result, str)
+
+    def test_contains_mod_function(self):
+        result = calc_scroll_x("bottom", 1000, 200, 50)
+        assert "mod" in result
+
+    def test_contains_speed_and_width(self):
+        result = calc_scroll_x("bottom", 1080, 300, 60)
+        assert "1080" in result
+        assert "60" in result
+        assert "300" in result
+
+
+# ── build_image_watermark_filter 测试 ──────────────────────────────────────
+
+
+class TestBuildImageWatermarkFilter:
+    def test_returns_tuple(self):
+        cfg = WatermarkConfig(mode="image", image_path="/wm.png")
+        result = build_image_watermark_filter("[in]", "/wm.png", 1080, 1920, "[out]", cfg)
+        assert isinstance(result, tuple)
+        assert len(result) == 2
+
+    def test_filter_contains_overlay(self):
+        cfg = WatermarkConfig(mode="image", image_path="/wm.png")
+        filter_str, inputs = build_image_watermark_filter("[in]", "/wm.png", 1080, 1920, "[out]", cfg)
+        assert "overlay" in filter_str
+
+    def test_filter_contains_scale(self):
+        cfg = WatermarkConfig(mode="image", image_path="/wm.png", scale=0.5)
+        filter_str, _ = build_image_watermark_filter("[in]", "/wm.png", 1080, 1920, "[out]", cfg)
+        assert "scale=" in filter_str
+
+    def test_full_opacity_no_alpha_filter(self):
+        cfg = WatermarkConfig(mode="image", image_path="/wm.png", opacity=1.0)
+        filter_str, _ = build_image_watermark_filter("[in]", "/wm.png", 1080, 1920, "[out]", cfg)
+        assert "colorchannelmixer" not in filter_str
+
+    def test_partial_opacity_has_alpha_filter(self):
+        cfg = WatermarkConfig(mode="image", image_path="/wm.png", opacity=0.5)
+        filter_str, _ = build_image_watermark_filter("[in]", "/wm.png", 1080, 1920, "[out]", cfg)
+        assert "colorchannelmixer" in filter_str
+        assert "aa=0.5" in filter_str
+
+    def test_input_args_contains_image_path(self):
+        cfg = WatermarkConfig(mode="image", image_path="/path/to/wm.png")
+        _, inputs = build_image_watermark_filter("[in]", "/path/to/wm.png", 1080, 1920, "[out]", cfg)
+        assert inputs == ["-i", "/path/to/wm.png"]
+
+    def test_scroll_mode_has_t_variable(self):
+        cfg = WatermarkConfig(mode="image", image_path="/wm.png", scroll=True, scroll_speed=50)
+        filter_str, _ = build_image_watermark_filter("[in]", "/wm.png", 1080, 1920, "[out]", cfg)
+        assert "t" in filter_str
+
+    def test_output_label_appears(self):
+        cfg = WatermarkConfig(mode="image", image_path="/wm.png")
+        filter_str, _ = build_image_watermark_filter("[in]", "/wm.png", 1080, 1920, "[final]", cfg)
+        assert "[final]" in filter_str
+
+
+# ── build_text_watermark_filter 测试 ───────────────────────────────────────
+
+
+class TestBuildTextWatermarkFilter:
+    def test_returns_string(self):
+        cfg = WatermarkConfig(mode="text", text="hello")
+        result = build_text_watermark_filter("[in]", "[out]", cfg, 1080, 1920)
+        assert isinstance(result, str)
+
+    def test_contains_drawtext(self):
+        cfg = WatermarkConfig(mode="text", text="hello")
+        result = build_text_watermark_filter("[in]", "[out]", cfg, 1080, 1920)
+        assert "drawtext=" in result
+
+    def test_contains_text_content(self):
+        cfg = WatermarkConfig(mode="text", text="watermark_test")
+        result = build_text_watermark_filter("[in]", "[out]", cfg, 1080, 1920)
+        assert "watermark_test" in result
+
+    def test_contains_font_size(self):
+        cfg = WatermarkConfig(mode="text", text="hi", font_size=48)
+        result = build_text_watermark_filter("[in]", "[out]", cfg, 1080, 1920)
+        assert "fontsize=48" in result
+
+    def test_contains_font_color(self):
+        cfg = WatermarkConfig(mode="text", text="hi", font_color="red")
+        result = build_text_watermark_filter("[in]", "[out]", cfg, 1080, 1920)
+        assert "fontcolor=red" in result
+
+    def test_font_path_included_when_set(self):
+        cfg = WatermarkConfig(mode="text", text="hi", font_path="/fonts/a.ttf")
+        result = build_text_watermark_filter("[in]", "[out]", cfg, 1080, 1920)
+        assert "fontfile=" in result
+        assert "a.ttf" in result
+
+    def test_font_path_not_included_when_empty(self):
+        cfg = WatermarkConfig(mode="text", text="hi")
+        result = build_text_watermark_filter("[in]", "[out]", cfg, 1080, 1920)
+        assert "fontfile=" not in result
+
+    def test_scroll_mode_has_t_variable(self):
+        cfg = WatermarkConfig(mode="text", text="hello", scroll=True, scroll_speed=30)
+        result = build_text_watermark_filter("[in]", "[out]", cfg, 1080, 1920)
+        assert "t" in result
+
+    def test_no_scroll_uses_fixed_position(self):
+        cfg = WatermarkConfig(mode="text", text="hello", scroll=False)
+        result = build_text_watermark_filter("[in]", "[out]", cfg, 1080, 1920)
+        assert "x=" in result
+        # 非滚动模式 x= 后面应该是数字，不是表达式
+        # 找 x= 后的第一个字符
+        import re
+
+        match = re.search(r"x=(\d+)", result)
+        assert match is not None
+
+    def test_output_label_appears(self):
+        cfg = WatermarkConfig(mode="text", text="hi")
+        result = build_text_watermark_filter("[in]", "[text_out]", cfg, 1080, 1920)
+        assert "[text_out]" in result
+
+    def test_special_chars_escaped(self):
+        cfg = WatermarkConfig(mode="text", text="hello:world")
+        result = build_text_watermark_filter("[in]", "[out]", cfg, 1080, 1920)
+        # 冒号应该被转义
+        assert "hello\\:world" in result or "hello\\\\\\:world" in result or "hello\\:" in result
+
+
+# ── 工具函数测试 ────────────────────────────────────────────────────────────
+
+
+class TestUtils:
+    def test_get_position_names_returns_nine(self):
+        names = get_position_names()
+        assert len(names) == 9
+
+    def test_get_position_names_all_valid(self):
+        names = get_position_names()
+        for name in names:
+            assert name in VALID_POSITIONS
+
+    def test_get_position_display_name_valid(self):
+        assert get_position_display_name("top_left") == "左上"
+        assert get_position_display_name("bottom_right") == "右下"
+
+    def test_get_position_display_name_invalid(self):
+        assert get_position_display_name("invalid") == "invalid"