"""贴纸叠加引擎 — 基于 FFmpeg overlay + drawtext 实现图片/文字贴纸. 支持能力: - 图片贴纸(PNG/GIF):位置、大小、透明度、时间范围、淡入淡出 - 文字贴纸(花字):字体、颜色、描边、阴影、位置、时间范围、动画 - 9宫格位置 + 自由坐标(像素或百分比) - 多贴纸叠加,按 z_index 排序 - 降级策略:素材不存在/无效时自动跳过,不阻断渲染 """ from __future__ import annotations import logging from dataclasses import dataclass, field from pathlib import Path from typing import Any logger = logging.getLogger(__name__) # ── 预设贴纸分类 ────────────────────────────────────────────────────────────── # 预设贴纸分类(仅用于前端展示,后端不依赖具体素材) STICKER_CATEGORIES = [ ("emoji", "表情包"), ("text", "文字花字"), ("decoration", "装饰"), ("arrow", "箭头指示"), ("frame", "边框"), ] # 9宫格位置映射 POSITION_PRESETS = { "top_left": (0.05, 0.05), "top_center": (0.5, 0.05), "top_right": (0.95, 0.05), "center_left": (0.05, 0.5), "center": (0.5, 0.5), "center_right": (0.95, 0.5), "bottom_left": (0.05, 0.95), "bottom_center": (0.5, 0.95), "bottom_right": (0.95, 0.95), } # ── 数据模型 ────────────────────────────────────────────────────────────────── @dataclass class ImageStickerConfig: """图片贴纸配置.""" enabled: bool = False type: str = "image" # image / text # 位置 position: str = "top_right" # 9宫格预设 x: float | None = None # 自定义x(像素或百分比) y: float | None = None # 自定义y x_unit: str = "percent" # pixel / percent y_unit: str = "percent" # 大小 scale: float = 1.0 # 缩放比例(相对于原始大小) width: int | None = None # 指定宽度(像素) height: int | None = None # 指定高度(像素) # 透明度 opacity: float = 1.0 # 0.0~1.0 # 时间范围 start_time: float = 0.0 duration: float = 0.0 # 0 表示持续到结束 # 动画 fade_in: float = 0.0 # 淡入时长(秒) fade_out: float = 0.0 # 淡出时长 # 层级 z_index: int = 10 # 素材 image_url: str = "" # 图片URL或本地路径 preset_id: str = "" # 预设贴纸ID @dataclass class TextStickerConfig: """文字贴纸配置.""" enabled: bool = False type: str = "text" text: str = "" # 字体 font_size: int = 36 font_color: str = "#FFFFFF" font_family: str = "sans" # 描边 stroke_color: str = "#000000" stroke_width: int = 2 # 阴影 shadow_color: str = "#000000" shadow_x: int = 2 shadow_y: int = 2 shadow_alpha: float = 0.5 # 位置 position: str = "center" x: float | None = None y: float | None = None x_unit: str = "percent" y_unit: str = "percent" # 时间范围 start_time: float = 0.0 duration: float = 0.0 # 动画 fade_in: float = 0.0 fade_out: float = 0.0 # 层级 z_index: int = 10 # 背景框 bg_color: str = "" # 空表示无背景 bg_padding: int = 8 bg_alpha: float = 0.8 bg_corner_radius: int = 8 @dataclass class StickerOverlayResult: """贴纸叠加结果.""" filter_str: str # 滤镜字符串 output_label: str # 输出标签 extra_inputs: list[str] = field(default_factory=list) # 额外的输入文件路径 # ── 贴纸引擎 ────────────────────────────────────────────────────────────────── class StickerEngine: """贴纸叠加引擎 — 生成 FFmpeg overlay / drawtext 滤镜链. 支持图片贴纸(overlay)和文字贴纸(drawtext)。 多贴纸按 z_index 排序依次叠加。 """ @staticmethod def _resolve_position( config: ImageStickerConfig | TextStickerConfig, canvas_w: int, canvas_h: int, sticker_w: int = 0, sticker_h: int = 0, ) -> tuple[float, float]: """解析贴纸位置(像素坐标). 优先级:自定义坐标 > 9宫格预设 """ # 先取预设的基准位置 if config.position in POSITION_PRESETS: px, py = POSITION_PRESETS[config.position] else: px, py = 0.5, 0.5 # 默认居中 # 自定义坐标覆盖 if config.x is not None: if config.x_unit == "percent": px = config.x / 100.0 else: px = config.x / canvas_w if canvas_w > 0 else 0.5 if config.y is not None: if config.y_unit == "percent": py = config.y / 100.0 else: py = config.y / canvas_h if canvas_h > 0 else 0.5 # 转换为像素坐标(考虑贴纸尺寸,使位置为贴纸中心点) x = px * canvas_w - sticker_w / 2 y = py * canvas_h - sticker_h / 2 # 钳制在画布内 x = max(0, min(x, canvas_w - sticker_w)) y = max(0, min(y, canvas_h - sticker_h)) return x, y @staticmethod def _build_overlay_filter( sticker: ImageStickerConfig, sticker_idx: int, input_label: str, output_label: str, canvas_w: int, canvas_h: int, ) -> str: """构建单个图片贴纸的 overlay 滤镜. Args: sticker: 贴纸配置 sticker_idx: 贴纸索引(用于生成滤镜标签) input_label: 输入视频标签(如 "[base]") output_label: 输出视频标签 canvas_w: 画布宽度 canvas_h: 画布高度 Returns: FFmpeg 滤镜字符串 """ sticker_label = f"sticker_{sticker_idx}_scaled" # 1. 贴纸缩放预处理 scale_parts = [] if sticker.width and sticker.height: scale_parts.append(f"scale={sticker.width}:{sticker.height}") elif sticker.scale != 1.0: # 按比例缩放 scale_parts.append(f"scale=iw*{sticker.scale}:ih*{sticker.scale}") # 透明度调整 if sticker.opacity < 1.0: scale_parts.append(f"colorchannelmixer=aa={sticker.opacity}") # 淡入淡出 fade_parts = [] if sticker.fade_in > 0: fade_parts.append(f"fade=in:st={sticker.start_time}:d={sticker.fade_in}:alpha=1") if sticker.fade_out > 0 and sticker.duration > 0: fade_out_start = sticker.start_time + sticker.duration - sticker.fade_out fade_parts.append(f"fade=out:st={max(0, fade_out_start)}:d={sticker.fade_out}:alpha=1") pre_filters = scale_parts + fade_parts # 2. overlay 位置 # 先估算贴纸尺寸(假设原始尺寸 ~ canvas_w * 0.3) est_w = int(canvas_w * 0.3 * sticker.scale) if not sticker.width else sticker.width est_h = int(canvas_h * 0.3 * sticker.scale) if not sticker.height else sticker.height pos_x, pos_y = StickerEngine._resolve_position(sticker, canvas_w, canvas_h, est_w, est_h) # 3. enable 表达式(时间范围) enable_expr = "" if sticker.duration > 0: enable_expr = f":enable='between(t,{sticker.start_time},{sticker.start_time + sticker.duration})'" # 组合滤镜 filter_parts: list[str] = [] # 贴纸预处理 if pre_filters: filter_parts.append(f"[{sticker_idx + 1}:v]{','.join(pre_filters)}[{sticker_label}]") sticker_source = f"[{sticker_label}]" else: sticker_source = f"[{sticker_idx + 1}:v]" # overlay 合成 filter_parts.append(f"{input_label}{sticker_source}overlay={pos_x:.0f}:{pos_y:.0f}{enable_expr}{output_label}") return ";".join(filter_parts) @staticmethod def _build_drawtext_filter( sticker: TextStickerConfig, input_label: str, output_label: str, canvas_w: int, canvas_h: int, ) -> str: """构建单个文字贴纸的 drawtext 滤镜. Args: sticker: 文字贴纸配置 input_label: 输入视频标签 output_label: 输出视频标签 canvas_w: 画布宽度 canvas_h: 画布高度 Returns: FFmpeg 滤镜字符串 """ if not sticker.text: return f"{input_label}copy{output_label}" # 估算文字尺寸(粗略) est_w = len(sticker.text) * sticker.font_size * 0.6 est_h = sticker.font_size * 1.4 pos_x, pos_y = StickerEngine._resolve_position(sticker, canvas_w, canvas_h, int(est_w), int(est_h)) drawtext_params: list[str] = [] # 文字内容(转义特殊字符) escaped_text = sticker.text.replace(":", "\\:").replace("'", "\\'") drawtext_params.append(f"text='{escaped_text}'") # 字体 drawtext_params.append(f"fontsize={sticker.font_size}") drawtext_params.append(f"fontcolor={sticker.font_color}") # 描边 if sticker.stroke_width > 0: drawtext_params.append(f"borderw={sticker.stroke_width}") drawtext_params.append(f"bordercolor={sticker.stroke_color}") # 阴影 if sticker.shadow_alpha > 0: drawtext_params.append(f"shadowx={sticker.shadow_x}") drawtext_params.append(f"shadowy={sticker.shadow_y}") drawtext_params.append(f"shadowcolor={sticker.shadow_color}@{sticker.shadow_alpha}") # 位置 drawtext_params.append(f"x={pos_x:.0f}") drawtext_params.append(f"y={pos_y:.0f}") # 时间范围 if sticker.duration > 0: drawtext_params.append(f"enable='between(t,{sticker.start_time},{sticker.start_time + sticker.duration})'") # 淡入淡出(drawtext 没有直接的淡入淡出,用 alpha 表达式模拟) if sticker.fade_in > 0 or sticker.fade_out > 0: alpha_expr = "1" parts: list[str] = [] if sticker.fade_in > 0: parts.append( f"if(lt(t,{sticker.start_time + sticker.fade_in})," f"(t-{sticker.start_time})/{sticker.fade_in},1)" ) if sticker.fade_out > 0 and sticker.duration > 0: fade_out_start = sticker.start_time + sticker.duration - sticker.fade_out parts.append( f"if(gt(t,{fade_out_start})," f"({sticker.start_time + sticker.duration}-t)/{sticker.fade_out},1)" ) if parts: alpha_expr = "*".join(parts) drawtext_params.append(f"alpha='{alpha_expr}'") filter_str = f"{input_label}drawtext={':'.join(drawtext_params)}{output_label}" return filter_str @classmethod def build_sticker_chain( cls, stickers: list[dict[str, Any]], input_label: str, output_label: str, canvas_w: int, canvas_h: int, ) -> StickerOverlayResult: """构建多贴纸叠加滤镜链. Args: stickers: 贴纸配置列表 input_label: 初始输入标签 output_label: 最终输出标签 canvas_w: 画布宽度 canvas_h: 画布高度 Returns: StickerOverlayResult,包含滤镜字符串、输出标签、额外输入 """ if not stickers: return StickerOverlayResult( filter_str=f"{input_label}copy{output_label}", output_label=output_label, extra_inputs=[], ) # 解析配置 parsed_stickers: list[tuple[int, ImageStickerConfig | TextStickerConfig]] = [] image_stickers: list[ImageStickerConfig] = [] image_paths: list[str] = [] for _, s in enumerate(stickers): try: sticker_type = s.get("type", "image") z = int(s.get("z_index", 10)) if sticker_type == "text": config = TextStickerConfig( enabled=True, text=str(s.get("text", "")), font_size=int(s.get("font_size", 36)), font_color=str(s.get("font_color", "#FFFFFF")), stroke_color=str(s.get("stroke_color", "#000000")), stroke_width=int(s.get("stroke_width", 2)), shadow_x=int(s.get("shadow_x", 2)), shadow_y=int(s.get("shadow_y", 2)), shadow_alpha=float(s.get("shadow_alpha", 0.5)), position=str(s.get("position", "center")), x=cls._safe_float(s.get("x")), y=cls._safe_float(s.get("y")), x_unit=str(s.get("x_unit", "percent")), y_unit=str(s.get("y_unit", "percent")), start_time=float(s.get("start_time", 0)), duration=float(s.get("duration", 0)), fade_in=float(s.get("fade_in", 0)), fade_out=float(s.get("fade_out", 0)), z_index=z, bg_color=str(s.get("bg_color", "")), bg_padding=int(s.get("bg_padding", 8)), bg_alpha=float(s.get("bg_alpha", 0.8)), bg_corner_radius=int(s.get("bg_corner_radius", 8)), ) parsed_stickers.append((z, config)) else: # 图片贴纸 — 安全校验:区分本地路径和URL image_path = s.get("image_path", "") image_url = s.get("image_url", "") safe_image_path: Path | None = None if image_path: # 本地路径:路径遍历防护 from video_processing.path_security import is_in_allowed_dirs try: p = Path(image_path) if not p.exists(): logger.warning("贴纸素材不存在,跳过: %s", image_path[:80]) continue if not is_in_allowed_dirs(p): logger.warning("贴纸路径不在允许目录内,拒绝: %s", image_path[:80]) continue safe_image_path = p.resolve() except Exception as e: logger.warning("贴纸路径校验失败,跳过: %s error=%s", image_path[:80], e) continue elif image_url: # URL:SSRF 安全校验(暂不自动下载,仅校验安全性) from video_processing.url_security import ( UrlSecurityError, validate_url_safety, ) try: validate_url_safety(image_url, purpose="sticker_image") except UrlSecurityError as e: logger.warning("贴纸URL安全校验失败,跳过: %s error=%s", image_url[:80], e) continue # URL 类型暂不支持自动下载,跳过 logger.info("贴纸URL类型暂不支持自动下载,跳过: %s", image_url[:80]) continue else: logger.warning("贴纸缺少 image_path 和 image_url,跳过") continue if safe_image_path is None: continue config = ImageStickerConfig( enabled=True, position=str(s.get("position", "top_right")), x=cls._safe_float(s.get("x")), y=cls._safe_float(s.get("y")), x_unit=str(s.get("x_unit", "percent")), y_unit=str(s.get("y_unit", "percent")), scale=float(s.get("scale", 1.0)), width=int(s["width"]) if s.get("width") else None, height=int(s["height"]) if s.get("height") else None, opacity=max(0.0, min(1.0, float(s.get("opacity", 1.0)))), start_time=float(s.get("start_time", 0)), duration=float(s.get("duration", 0)), fade_in=float(s.get("fade_in", 0)), fade_out=float(s.get("fade_out", 0)), z_index=z, image_url=image_url, ) parsed_stickers.append((z, config)) image_stickers.append(config) image_paths.append(str(safe_image_path)) except Exception as e: logger.warning("贴纸配置解析失败,跳过: %s", e) continue if not parsed_stickers: return StickerOverlayResult( filter_str=f"{input_label}copy{output_label}", output_label=output_label, extra_inputs=[], ) # 按 z_index 排序 parsed_stickers.sort(key=lambda x: x[0]) # 构建滤镜链 filter_parts: list[str] = [] current_label = input_label img_idx = 0 # 图片贴纸的输入索引偏移 for idx, (_, sticker) in enumerate(parsed_stickers): next_label = f"sticker_{idx}_out" if idx < len(parsed_stickers) - 1 else output_label if isinstance(sticker, ImageStickerConfig): # 图片贴纸:使用额外的输入(输入索引 = 1 + img_idx,0 是主视频) # 注意:实际输入索引需要调用方根据输入列表确定 # 这里我们按 image_stickers 的顺序分配索引 # 主输入是 [0:v],贴纸输入从 [1:v] 开始 single_filter = cls._build_single_image_sticker( sticker=sticker, sticker_input_idx=img_idx + 1, # +1 因为 0 是主视频 input_label=current_label, output_label=next_label, canvas_w=canvas_w, canvas_h=canvas_h, ) filter_parts.append(single_filter) img_idx += 1 else: # 文字贴纸:drawtext,不需要额外输入 single_filter = cls._build_drawtext_filter( sticker, # type: ignore current_label, next_label, canvas_w, canvas_h, ) filter_parts.append(single_filter) current_label = next_label return StickerOverlayResult( filter_str=";".join(filter_parts), output_label=output_label, extra_inputs=image_paths, ) @classmethod def _build_single_image_sticker( cls, sticker: ImageStickerConfig, sticker_input_idx: int, input_label: str, output_label: str, canvas_w: int, canvas_h: int, ) -> str: """构建单个图片贴纸的完整滤镜(预处理 + overlay). Args: sticker: 贴纸配置 sticker_input_idx: 贴纸在 FFmpeg 输入中的索引 input_label: 输入视频标签 output_label: 输出标签 canvas_w: 画布宽 canvas_h: 画布高 """ scaled_label = f"sticker_s{sticker_input_idx}" # 预处理滤镜(缩放 + 透明度 + 淡入淡出) pre_filters: list[str] = [] # 缩放 if sticker.width and sticker.height: pre_filters.append(f"scale={sticker.width}:{sticker.height}") elif sticker.scale != 1.0: pre_filters.append(f"scale=iw*{sticker.scale}:ih*{sticker.scale}") # 透明度 if sticker.opacity < 1.0: pre_filters.append(f"format=rgba,colorchannelmixer=aa={sticker.opacity}") # 淡入淡出(使用 fade 的 alpha 模式) fade_filters: list[str] = [] if sticker.fade_in > 0: fade_filters.append(f"fade=in:st={sticker.start_time}:d={sticker.fade_in}:alpha=1") if sticker.fade_out > 0 and sticker.duration > 0: fade_out_start = sticker.start_time + sticker.duration - sticker.fade_out if fade_out_start > 0: fade_filters.append(f"fade=out:st={fade_out_start}:d={sticker.fade_out}:alpha=1") # 估算贴纸尺寸用于位置计算 est_w = int(canvas_w * 0.3 * sticker.scale) if not sticker.width else sticker.width est_h = int(canvas_h * 0.3 * sticker.scale) if not sticker.height else sticker.height pos_x, pos_y = cls._resolve_position(sticker, canvas_w, canvas_h, est_w, est_h) # enable 表达式 enable_expr = "" if sticker.duration > 0: enable_expr = f":enable='between(t,{sticker.start_time},{sticker.start_time + sticker.duration})'" parts: list[str] = [] # 贴纸预处理 all_pre = pre_filters + fade_filters if all_pre: parts.append(f"[{sticker_input_idx}:v]{','.join(all_pre)}[{scaled_label}]") sticker_source = f"[{scaled_label}]" else: sticker_source = f"[{sticker_input_idx}:v]" # overlay 合成 parts.append(f"{input_label}{sticker_source}overlay={pos_x:.0f}:{pos_y:.0f}{enable_expr}{output_label}") return ";".join(parts) @staticmethod def _safe_float(val: Any) -> float | None: """安全转换 float.""" if val is None: return None try: return float(val) except (ValueError, TypeError): return None # ── 便捷函数 ────────────────────────────────────────────────────────────────── def parse_stickers_from_config(config: dict[str, Any] | None) -> list[dict[str, Any]]: """从 plan.config.stickers 解析贴纸列表.""" if not config: return [] stickers = config.get("stickers", []) if not isinstance(stickers, list): return [] return stickers def get_sticker_categories() -> list[tuple[str, str]]: """获取贴纸分类列表.""" return list(STICKER_CATEGORIES)