app/operations/crop.py

from __future__ import annotations

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

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
from app.services.validator import positive_int


def _crop_filter(params: dict[str, Any]) -> tuple[str, dict[str, int]]:
    width = positive_int(params, "width", 640)
    height = positive_int(params, "height", 360)
    x = int(params.get("x", 0))
    y = int(params.get("y", 0))
    if x < 0 or y < 0:
        from app.core.exceptions import InputError

        raise InputError("Crop x and y must not be negative")
    return f"crop={width}:{height}:{x}:{y}", {"width": width, "height": height, "x": x, "y": y}


async def crop_video(
    ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
    require_inputs(inputs)
    vf, metadata = _crop_filter(params)
    output = output_path(output_dir, "cropped", "mp4")
    return await execute(
        ffmpeg,
        ["-i", inputs[0].temp_path, "-vf", vf, *video_codecs("mp4"), output],
        output,
        "video.crop",
        metadata,
    )


async def crop_image(
    ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
    require_inputs(inputs)
    vf, metadata = _crop_filter(params)
    extension = format_param(params, "png", IMAGE_FORMATS)
    output = output_path(output_dir, "cropped", extension)
    return await execute(
        ffmpeg,
        ["-i", inputs[0].temp_path, "-vf", vf, "-frames:v", "1", output],
        output,
        "image.crop",
        metadata,
    )