app/operations/convert.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 (
    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})