Spaces:
Sleeping
Sleeping
File size: 3,548 Bytes
fba6023 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 | 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},
)
|