"""贴纸叠加引擎 — 基于 FFmpeg overlay + drawtext 实现图片/文字贴纸. 支持能力： - 图片贴纸（PNG/GIF）：位置、大小、透明度、时间范围、淡入淡出 - 文字贴纸（花字）：字体、颜色、描边、阴影、位置、时间范围、动画 - 9宫格位置 + 自由坐标（像素或百分比） - 多贴纸叠加，按 z_index 排序 - 降级策略：素材不存在/无效时自动跳过，不阻断渲染 """ from __future__ import annotations import logging from pathlib import Path from typing import Any from packages.domain.sticker_config import ( POSITION_PRESETS, STICKER_CATEGORIES, ImageStickerConfig, StickerOverlayResult, TextStickerConfig, ) from packages.domain.sticker_config import get_sticker_categories as _get_sticker_categories_base # noqa: F401 向后兼容导出 from packages.domain.sticker_config import parse_stickers_from_config as _parse_stickers_base from packages.domain.sticker_config import ( resolve_sticker_position, ) logger = logging.getLogger(__name__) # ── 贴纸引擎 ────────────────────────────────────────────────────────────────── 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]: """解析贴纸位置（像素坐标）（转发到 sticker_config 模块）.""" return resolve_sticker_position( config.position, config.x, config.y, config.x_unit, config.y_unit, canvas_w, canvas_h, sticker_w, sticker_h, ) @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 # ── 便捷函数（薄包装，转发到 sticker_config 模块） ──────────────────────────── def parse_stickers_from_config(config: dict[str, Any] | None) -> list[dict[str, Any]]: """从 plan.config.stickers 解析贴纸列表（薄包装）.""" return _parse_stickers_base(config) def get_sticker_categories() -> list[tuple[str, str]]: """获取贴纸分类列表（薄包装）.""" return _get_sticker_categories_base()