xiaoxia-saas/apps/worker/video_processing/sticker_engine.py

"""贴纸叠加引擎 — 基于 FFmpeg overlay + drawtext 实现图片/文字贴纸.

支持能力：
- 图片贴纸（PNG/GIF）：位置、大小、透明度、时间范围、淡入淡出
- 文字贴纸（花字）：字体、颜色、描边、阴影、位置、时间范围、动画
- 9宫格位置 + 自由坐标（像素或百分比）
- 多贴纸叠加，按 z_index 排序
- 降级策略：素材不存在/无效时自动跳过，不阻断渲染
"""

from __future__ import annotations

import logging
from pathlib import Path
from typing import Any

# isort: off
from packages.domain.sticker_config import (
    POSITION_PRESETS,  # noqa: F401
    STICKER_CATEGORIES,  # noqa: F401
    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 (
    # isort: on
    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()