diff --git a/apps/worker/video_processing/watermark_engine.py b/apps/worker/video_processing/watermark_engine.py index a1a1a83fd..c464ebb6e 100755 --- a/apps/worker/video_processing/watermark_engine.py +++ b/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) diff --git a/packages/domain/watermark_config.py b/packages/domain/watermark_config.py new file mode 100755 index 000000000..5aa05b47d --- /dev/null +++ b/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) diff --git a/tests/unit/test_watermark_config.py b/tests/unit/test_watermark_config.py new file mode 100755 index 000000000..bce7866a7 --- /dev/null +++ b/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"