"""水印配置领域模型 — 纯逻辑,无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 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)