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

"""水印引擎 — 基于 FFmpeg overlay 滤镜的水印叠加.

支持：
- 图片水印（PNG/logo）
- 文字水印（drawtext）
- 9宫格位置 + 边距配置
- 透明度/大小缩放
- 滚动水印（跑马灯）

注：核心领域模型已抽离到 packages/domain/watermark_config.py，
本模块保留薄包装层，确保向后兼容。
"""

from __future__ import annotations

import logging
from typing import Any

from packages.domain.watermark_config import (  # noqa: F401 — 向后兼容
    WATERMARK_POSITIONS,
    WatermarkConfig,
    build_image_watermark_filter as _build_image_watermark_filter,
    build_text_watermark_filter as _build_text_watermark_filter,
    calc_position as _calc_position_base,
    calc_scroll_x as _calc_scroll_x_base,
)

logger = logging.getLogger(__name__)


class WatermarkEngine:
    """水印引擎 — 生成 FFmpeg 水印滤镜.

    薄包装层，实际逻辑委托给 packages.domain.watermark_config。
    """

    @staticmethod
    def calc_position(
        position: str,
        output_width: int,
        output_height: int,
        wm_width: int,
        wm_height: int,
        margin_x: int,
        margin_y: int,
    ) -> tuple[int, int]:
        """根据9宫格位置计算水印坐标 (x, y).

        坐标系：左上角为 (0, 0)
        """
        return _calc_position_base(position, output_width, output_height, wm_width, wm_height, margin_x, margin_y)

    @staticmethod
    def calc_scroll_x(position: str, output_width: int, wm_width: int, speed: int) -> str:
        """生成滚动水印的 x 坐标表达式.

        从右向左滚动（跑马灯效果）
        """
        return _calc_scroll_x_base(position, output_width, wm_width, speed)

    @staticmethod
    def build_image_watermark_filter(
        input_video_label: str,
        wm_image_path: str,
        output_width: int,
        output_height: int,
        output_label: str,
        config: WatermarkConfig,
    ) -> tuple[str, list[str]]:
        """构建图片水印滤镜链.

        Args:
            input_video_label: 输入视频标签，如 "[final_video]"
            wm_image_path: 水印图片本地路径
            output_width: 输出视频宽度
            output_height: 输出视频高度
            output_label: 输出标签
            config: 水印配置

        Returns:
            (filter_complex_str, input_args_list)
            input_args 是 ["-i", wm_image_path] 格式
        """
        return _build_image_watermark_filter(
            input_video_label,
            wm_image_path,
            output_width,
            output_height,
            output_label,
            config,
        )

    @staticmethod
    def build_text_watermark_filter(
        input_video_label: str,
        output_label: str,
        config: WatermarkConfig,
        output_width: int,
        output_height: int,
    ) -> str:
        """构建文字水印滤镜（drawtext）.

        Args:
            input_video_label: 输入视频标签
            output_label: 输出标签
            config: 水印配置
            output_width: 输出宽度
            output_height: 输出高度

        Returns:
            FFmpeg filter 字符串
        """
        return _build_text_watermark_filter(input_video_label, output_label, config, output_width, output_height)