"""贴纸引擎纯逻辑模块. 所有函数均为纯函数,不调用 FFmpeg、不操作文件。 便于单元测试,也方便被其他模块复用。 """ from __future__ import annotations from typing import Any, Optional # ── 安全类型转换 ────────────────────────────────────────────────────────────── def safe_float(val: Any) -> Optional[float]: """安全转换为 float. Args: val: 任意值 Returns: float 值,失败返回 None """ if val is None: return None try: result = float(val) if result != result: # NaN check return None return result except (ValueError, TypeError): return None def safe_int(val: Any, default: int = 0) -> int: """安全转换为 int. Args: val: 任意值 default: 默认值 Returns: int 值,失败返回默认值 """ if val is None: return default try: result = int(float(val)) return result except (ValueError, TypeError): return default def safe_bool(val: Any) -> bool: """安全转换为 bool. Args: val: 任意值 Returns: bool 值 """ if isinstance(val, bool): return val if val is None: return False if isinstance(val, str): return val.lower() in ("true", "1", "yes", "on") return bool(val) # ── 尺寸估算 ────────────────────────────────────────────────────────────────── def estimate_sticker_size( canvas_w: int, canvas_h: int, scale: float = 1.0, fixed_width: Optional[int] = None, fixed_height: Optional[int] = None, ) -> tuple[int, int]: """估算贴纸尺寸. 如果指定了固定宽高,直接使用;否则按画布的 30% * scale 估算。 Args: canvas_w: 画布宽度 canvas_h: 画布高度 scale: 缩放比例 fixed_width: 固定宽度(可选) fixed_height: 固定高度(可选) Returns: (估算宽度, 估算高度) """ if fixed_width and fixed_height: return (fixed_width, fixed_height) base_ratio = 0.3 est_w = int(canvas_w * base_ratio * scale) if not fixed_width else fixed_width est_h = int(canvas_h * base_ratio * scale) if not fixed_height else fixed_height return (max(1, est_w), max(1, est_h)) def estimate_text_size( text: str, font_size: int, ) -> tuple[int, int]: """估算文字贴纸尺寸. 粗略估算:宽度 = 字数 * 字号 * 0.6,高度 = 字号 * 1.4 Args: text: 文字内容 font_size: 字号 Returns: (估算宽度, 估算高度) """ if not text: return (0, 0) est_w = int(len(text) * font_size * 0.6) est_h = int(font_size * 1.4) return (max(1, est_w), max(1, est_h)) # ── 时间计算 ────────────────────────────────────────────────────────────────── def calculate_fade_out_start( start_time: float, duration: float, fade_out: float, ) -> float: """计算淡出开始时间. Args: start_time: 开始时间(秒) duration: 持续时长(秒) fade_out: 淡出时长(秒) Returns: 淡出开始时间(秒),最小为 0 """ if fade_out <= 0 or duration <= 0: return 0.0 fade_start = start_time + duration - fade_out return max(0.0, fade_start) def calculate_end_time(start_time: float, duration: float) -> float: """计算结束时间. Args: start_time: 开始时间(秒) duration: 持续时长(秒) Returns: 结束时间(秒) """ if duration <= 0: return start_time return start_time + duration def has_time_range(duration: float) -> bool: """是否有时间范围限制. Args: duration: 持续时长(秒) Returns: duration > 0 时返回 True """ return duration > 0 # ── 滤镜组件构建 ────────────────────────────────────────────────────────────── def build_scale_filter( width: Optional[int] = None, height: Optional[int] = None, scale: float = 1.0, ) -> Optional[str]: """构建缩放滤镜. 优先使用固定宽高,否则按比例缩放。 scale=1.0 且无固定尺寸时返回 None。 Args: width: 固定宽度(可选) height: 固定高度(可选) scale: 缩放比例 Returns: scale 滤镜字符串,不需要缩放时返回 None """ if width and height: return f"scale={width}:{height}" if scale != 1.0: return f"scale=iw*{scale}:ih*{scale}" return None def build_opacity_filter(opacity: float) -> Optional[str]: """构建透明度滤镜. Args: opacity: 不透明度 0.0~1.0 Returns: colorchannelmixer 滤镜字符串,完全不透明时返回 None """ if opacity >= 1.0: return None safe_opacity = max(0.0, min(1.0, opacity)) return f"colorchannelmixer=aa={safe_opacity}" def build_image_fade_filters( start_time: float, duration: float, fade_in: float = 0.0, fade_out: float = 0.0, ) -> list[str]: """构建图片贴纸淡入淡出滤镜列表. 使用 FFmpeg fade 滤镜(alpha 通道)。 Args: start_time: 开始时间(秒) duration: 持续时长(秒) fade_in: 淡入时长(秒) fade_out: 淡出时长(秒) Returns: fade 滤镜字符串列表 """ filters: list[str] = [] if fade_in > 0: filters.append(f"fade=in:st={start_time}:d={fade_in}:alpha=1") if fade_out > 0 and duration > 0: fade_out_start = calculate_fade_out_start(start_time, duration, fade_out) filters.append(f"fade=out:st={fade_out_start}:d={fade_out}:alpha=1") return filters def build_enable_expr( start_time: float, duration: float, ) -> str: """构建 enable 表达式(时间范围). Args: start_time: 开始时间(秒) duration: 持续时长(秒) Returns: enable 表达式字符串(包含开头的冒号),无时间限制时返回空字符串 """ if duration <= 0: return "" end_time = calculate_end_time(start_time, duration) return f":enable='between(t,{start_time},{end_time})'" # ── drawtext 文字贴纸相关 ───────────────────────────────────────────────────── def escape_drawtext_text(text: str) -> str: """转义 drawtext 中的特殊字符. 转义冒号和单引号。 Args: text: 原始文字 Returns: 转义后的文字 """ result = text.replace(":", "\\:") result = result.replace("'", "\\'") return result def build_drawtext_alpha_expr( start_time: float, duration: float, fade_in: float = 0.0, fade_out: float = 0.0, ) -> str: """构建 drawtext 的 alpha 淡入淡出表达式. drawtext 没有直接的 fade 滤镜,用 alpha 表达式模拟。 Args: start_time: 开始时间(秒) duration: 持续时长(秒) fade_in: 淡入时长(秒) fade_out: 淡出时长(秒) Returns: alpha 表达式字符串,无淡入淡出时返回 "1" """ parts: list[str] = [] if fade_in > 0: fade_in_end = start_time + fade_in parts.append(f"if(lt(t,{fade_in_end}),(t-{start_time})/{fade_in},1)") if fade_out > 0 and duration > 0: fade_out_start = calculate_fade_out_start(start_time, duration, fade_out) end_time = calculate_end_time(start_time, duration) parts.append(f"if(gt(t,{fade_out_start}),({end_time}-t)/{fade_out},1)") if not parts: return "1" return "*".join(parts) def build_stroke_params( stroke_width: int = 0, stroke_color: str = "black", ) -> list[str]: """构建 drawtext 描边参数. Args: stroke_width: 描边宽度(0 表示无描边) stroke_color: 描边颜色 Returns: 描边参数列表 """ if stroke_width <= 0: return [] return [ f"borderw={stroke_width}", f"bordercolor={stroke_color}", ] def build_shadow_params( shadow_alpha: float = 0.0, shadow_x: int = 2, shadow_y: int = 2, shadow_color: str = "black", ) -> list[str]: """构建 drawtext 阴影参数. Args: shadow_alpha: 阴影透明度(0 表示无阴影) shadow_x: 阴影 X 偏移 shadow_y: 阴影 Y 偏移 shadow_color: 阴影颜色 Returns: 阴影参数列表 """ if shadow_alpha <= 0: return [] safe_alpha = max(0.0, min(1.0, shadow_alpha)) return [ f"shadowx={shadow_x}", f"shadowy={shadow_y}", f"shadowcolor={shadow_color}@{safe_alpha}", ] # ── 贴纸排序与过滤 ──────────────────────────────────────────────────────────── def sort_stickers_by_z_index( stickers: list[dict[str, Any]], ) -> list[dict[str, Any]]: """按 z_index 排序贴纸. z_index 小的在底层,大的在上层。 相同 z_index 保持原顺序(稳定排序)。 Args: stickers: 贴纸配置列表 Returns: 排序后的贴纸列表 """ return sorted(stickers, key=lambda s: safe_int(s.get("z_index"), 10)) def filter_enabled_stickers( stickers: list[dict[str, Any]], ) -> list[dict[str, Any]]: """过滤出启用的贴纸. Args: stickers: 贴纸配置列表 Returns: 启用的贴纸列表 """ result = [] for s in stickers: enabled = s.get("enabled", True) if safe_bool(enabled): result.append(s) return result def count_sticker_types( stickers: list[dict[str, Any]], ) -> dict[str, int]: """统计各类型贴纸数量. Args: stickers: 贴纸配置列表 Returns: 类型计数字典 """ counts: dict[str, int] = {} for s in stickers: stype = s.get("type", "image") counts[stype] = counts.get(stype, 0) + 1 return counts # ── overlay 滤镜构建 ────────────────────────────────────────────────────────── def build_overlay_position( pos_x: float, pos_y: float, ) -> str: """构建 overlay 位置参数. Args: pos_x: X 坐标 pos_y: Y 坐标 Returns: overlay 位置字符串 "x:y" """ return f"{pos_x:.0f}:{pos_y:.0f}" def build_pre_filter_label(idx: int) -> str: """构建贴纸预处理后的标签名. Args: idx: 贴纸索引 Returns: 滤镜标签字符串 """ return f"sticker_{idx}_scaled" # ── 验证函数 ────────────────────────────────────────────────────────────────── def validate_image_sticker(sticker: dict[str, Any]) -> tuple[bool, list[str]]: """验证图片贴纸配置. Args: sticker: 贴纸配置字典 Returns: (是否合法, 错误信息列表) """ errors: list[str] = [] # 图片路径 image_path = sticker.get("image_path", "") if not image_path and not sticker.get("asset_id"): errors.append("图片贴纸需要 image_path 或 asset_id") # 透明度范围 opacity = safe_float(sticker.get("opacity", 1.0)) if opacity is not None and (opacity < 0 or opacity > 1): errors.append("opacity 必须在 0~1 之间") # 缩放比例 scale = safe_float(sticker.get("scale", 1.0)) if scale is not None and scale <= 0: errors.append("scale 必须大于 0") # 时间参数 duration = safe_float(sticker.get("duration", 0)) if duration is not None and duration < 0: errors.append("duration 不能为负数") start_time = safe_float(sticker.get("start_time", 0)) if start_time is not None and start_time < 0: errors.append("start_time 不能为负数") return (len(errors) == 0, errors) def validate_text_sticker(sticker: dict[str, Any]) -> tuple[bool, list[str]]: """验证文字贴纸配置. Args: sticker: 贴纸配置字典 Returns: (是否合法, 错误信息列表) """ errors: list[str] = [] # 文字内容 text = sticker.get("text", "") if not text: errors.append("文字贴纸需要 text 内容") # 字号 font_size = safe_int(sticker.get("font_size", 36)) if font_size <= 0: errors.append("font_size 必须大于 0") # 颜色 font_color = sticker.get("font_color", "white") if not font_color: errors.append("font_color 不能为空") # 时间参数 duration = safe_float(sticker.get("duration", 0)) if duration is not None and duration < 0: errors.append("duration 不能为负数") return (len(errors) == 0, errors)