Spaces:
Sleeping
Sleeping
File size: 1,915 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 | 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 (
AUDIO_FORMATS,
IMAGE_FORMATS,
VIDEO_FORMATS,
audio_codec,
execute,
format_param,
output_path,
require_inputs,
video_codecs,
)
from app.services.ffmpeg_service import FFmpegService
async def convert_video(
ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
require_inputs(inputs)
extension = format_param(params, "mp4", VIDEO_FORMATS)
output = output_path(output_dir, "converted", extension)
args = ["-i", inputs[0].temp_path, *video_codecs(extension), output]
return await execute(ffmpeg, args, output, "video.convert", {"format": extension})
async def convert_audio(
ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
require_inputs(inputs)
extension = format_param(params, "mp3", AUDIO_FORMATS)
output = output_path(output_dir, "converted", extension)
args = ["-i", inputs[0].temp_path, "-vn", *audio_codec(extension), output]
return await execute(ffmpeg, args, output, "audio.convert", {"format": extension})
async def convert_image(
ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
require_inputs(inputs)
extension = format_param(params, "png", IMAGE_FORMATS)
output = output_path(output_dir, "converted", extension)
args: list[str | Path] = ["-i", inputs[0].temp_path, "-frames:v", "1"]
if extension in {"jpg", "jpeg"}:
args += ["-q:v", str(params.get("quality", 2))]
args += [output]
return await execute(ffmpeg, args, output, "image.convert", {"format": extension})
|