app/operations/compress.py

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 (
    VIDEO_FORMATS,
    execute,
    format_param,
    output_path,
    require_inputs,
    video_codecs,
)
from app.services.ffmpeg_service import FFmpegService
from app.services.validator import bounded_number


async def compress_video(
    ffmpeg: FFmpegService,
    inputs: Sequence[InputMedia],
    params: dict[str, Any],
    output_dir: Path,
) -> OperationResult:
    require_inputs(inputs)
    extension = format_param(params, "mp4", VIDEO_FORMATS)
    try:
        crf = int(params.get("crf", 28))
    except (TypeError, ValueError) as exc:
        raise InputError("'crf' must be an integer") from exc
    if not 0 <= crf <= 51:
        raise InputError("'crf' must be between 0 and 51")
    preset = str(params.get("preset", "medium"))
    if preset not in {"ultrafast", "superfast", "veryfast", "faster", "fast", "medium", "slow"}:
        raise InputError("Unsupported x264 preset")
    output = output_path(output_dir, "compressed", extension)
    args: list[str | Path] = ["-i", inputs[0].temp_path]
    scale = params.get("max_width")
    if scale:
        try:
            max_width = int(scale)
        except (TypeError, ValueError) as exc:
            raise InputError("'max_width' must be an integer") from exc
        args += ["-vf", f"scale='min({max_width},iw)':-2"]
    args += video_codecs(extension, crf=crf, preset=preset)
    if params.get("bitrate"):
        bitrate = str(params["bitrate"])
        if not bitrate.replace("k", "").replace("M", "").isdigit():
            raise InputError("Invalid video bitrate")
        args += ["-b:v", bitrate]
    args += [output]
    return await execute(ffmpeg, args, output, "video.compress", {"crf": crf, "format": extension})


async def normalize_video(
    ffmpeg: FFmpegService,
    inputs: Sequence[InputMedia],
    params: dict[str, Any],
    output_dir: Path,
) -> OperationResult:
    require_inputs(inputs)
    output = output_path(output_dir, "normalized", "mp4")
    args: list[str | Path] = ["-i", inputs[0].temp_path]
    if inputs[0].metadata.get("audio_streams"):
        args += ["-af", "loudnorm=I=-16:LRA=11:TP=-1.5"]
    args += [*video_codecs("mp4", crf=23), output]
    return await execute(ffmpeg, args, output, "video.normalize")


async def normalize_audio(
    ffmpeg: FFmpegService,
    inputs: Sequence[InputMedia],
    params: dict[str, Any],
    output_dir: Path,
) -> OperationResult:
    require_inputs(inputs)
    target = bounded_number(params, "target_lufs", -16, -70, -5)
    output = output_path(output_dir, "normalized", "wav")
    args = [
        "-i",
        inputs[0].temp_path,
        "-af",
        f"loudnorm=I={target}:LRA=11:TP=-1.5",
        "-c:a",
        "pcm_s16le",
        output,
    ]
    return await execute(ffmpeg, args, output, "audio.normalize", {"target_lufs": target})