app/operations/watermark.py

from __future__ import annotations

from collections.abc import Sequence
from pathlib import Path
from typing import Any

from app.core.exceptions import InputError
from app.models.media import InputMedia, OperationResult
from app.operations.common import (
    IMAGE_FORMATS,
    execute,
    format_param,
    output_path,
    require_inputs,
    video_codecs,
)
from app.services.ffmpeg_service import FFmpegService


def _overlay(params: dict[str, Any]) -> tuple[str, dict[str, Any]]:
    position = str(params.get("position", "bottom-right"))
    positions = {
        "top-left": "10:10",
        "top-right": "W-w-10:10",
        "bottom-left": "10:H-h-10",
        "bottom-right": "W-w-10:H-h-10",
        "center": "(W-w)/2:(H-h)/2",
    }
    if position not in positions:
        raise InputError(f"position must be one of: {', '.join(positions)}")
    try:
        opacity = float(params.get("opacity", 1.0))
        scale = float(params.get("watermark_scale", 1.0))
    except (TypeError, ValueError) as exc:
        raise InputError("opacity and watermark_scale must be numbers") from exc
    if not 0 <= opacity <= 1 or not 0.01 <= scale <= 10:
        raise InputError("opacity must be 0..1 and watermark_scale must be 0.01..10")
    filter_graph = f"[1:v]format=rgba,colorchannelmixer=aa={opacity},scale=iw*{scale}:ih*{scale}[wm];[0:v][wm]overlay={positions[position]}[v]"
    return filter_graph, {"position": position, "opacity": opacity, "scale": scale}


async def watermark_video(
    ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
    require_inputs(inputs, 2)
    graph, metadata = _overlay(params)
    output = output_path(output_dir, "watermarked", "mp4")
    args = [
        "-i",
        inputs[0].temp_path,
        "-i",
        inputs[1].temp_path,
        "-filter_complex",
        graph,
        "-map",
        "[v]",
        "-map",
        "0:a?",
        *video_codecs("mp4"),
        output,
    ]
    return await execute(ffmpeg, args, output, "video.watermark", metadata)


async def overlay_video(
    ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
    result = await watermark_video(ffmpeg, inputs, params, output_dir)
    result.metadata["operation"] = "video.overlay"
    return result


async def watermark_image(
    ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
    require_inputs(inputs, 2)
    graph, metadata = _overlay(params)
    extension = format_param(params, "png", IMAGE_FORMATS)
    output = output_path(output_dir, "watermarked", extension)
    args = [
        "-i",
        inputs[0].temp_path,
        "-i",
        inputs[1].temp_path,
        "-filter_complex",
        graph,
        "-map",
        "[v]",
        "-frames:v",
        "1",
        output,
    ]
    return await execute(ffmpeg, args, output, "image.watermark", metadata)


async def overlay_image(
    ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
    result = await watermark_image(ffmpeg, inputs, params, output_dir)
    result.metadata["operation"] = "image.overlay"
    return result