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

"""贴纸引擎纯逻辑模块.

所有函数均为纯函数，不调用 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)