app/operations/resize.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,
    VIDEO_FORMATS,
    even,
    execute,
    format_param,
    output_path,
    require_inputs,
    video_codecs,
)
from app.services.ffmpeg_service import FFmpegService
from app.services.validator import positive_int


def _scale(params: dict[str, Any]) -> tuple[int, int, str]:
    width = even(positive_int(params, "width", 1280))
    height = even(positive_int(params, "height", 720))
    if width < 2 or height < 2:
        raise InputError("Output dimensions must be at least 2x2")
    fit = str(params.get("fit", "contain"))
    if fit == "fill":
        expression = f"scale={width}:{height}"
    elif fit == "cover":
        expression = (
            f"scale={width}:{height}:force_original_aspect_ratio=increase,crop={width}:{height}"
        )
    elif fit == "contain":
        expression = f"scale={width}:{height}:force_original_aspect_ratio=decrease,pad={width}:{height}:(ow-iw)/2:(oh-ih)/2"
    else:
        raise InputError("'fit' must be contain, cover, or fill")
    return width, height, expression


async def resize_video(
    ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
    require_inputs(inputs)
    width, height, expression = _scale(params)
    extension = format_param(params, "mp4", VIDEO_FORMATS)
    output = output_path(output_dir, "resized", extension)
    return await execute(
        ffmpeg,
        ["-i", inputs[0].temp_path, "-vf", expression, *video_codecs(extension), output],
        output,
        "video.resize",
        {"width": width, "height": height},
    )


async def scale_video(
    ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
    copied = {**params, "fit": params.get("fit", "fill")}
    result = await resize_video(ffmpeg, inputs, copied, output_dir)
    result.metadata["operation"] = "video.scale"
    return result


async def pad_video(
    ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
    require_inputs(inputs)
    width = even(positive_int(params, "width", 1920))
    height = even(positive_int(params, "height", 1080))
    color = str(params.get("color", "black"))
    if not color.replace("#", "").isalnum():
        raise InputError("Invalid pad color")
    output = output_path(output_dir, "padded", "mp4")
    vf = f"scale={width}:{height}:force_original_aspect_ratio=decrease,pad={width}:{height}:(ow-iw)/2:(oh-ih)/2:{color}"
    return await execute(
        ffmpeg,
        ["-i", inputs[0].temp_path, "-vf", vf, *video_codecs("mp4"), output],
        output,
        "video.pad",
        {"width": width, "height": height},
    )


async def resize_image(
    ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
    require_inputs(inputs)
    width, height, expression = _scale(params)
    extension = format_param(params, "png", IMAGE_FORMATS)
    output = output_path(output_dir, "resized", extension)
    return await execute(
        ffmpeg,
        ["-i", inputs[0].temp_path, "-vf", expression, "-frames:v", "1", output],
        output,
        "image.resize",
        {"width": width, "height": height},
    )