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, )