app/operations/common.py

from __future__ import annotations

import mimetypes
import re
from collections.abc import Sequence
from pathlib import Path
from typing import Any

from app.core.exceptions import InputError, ProcessingError
from app.models.media import InputMedia, OperationResult
from app.services.ffmpeg_service import FFmpegService

VIDEO_FORMATS = {"mp4", "mkv", "webm", "mov", "avi", "m4v", "mpeg", "ts"}
AUDIO_FORMATS = {"mp3", "wav", "aac", "m4a", "flac", "ogg", "opus"}
IMAGE_FORMATS = {"jpg", "jpeg", "png", "webp", "bmp", "tiff", "gif"}


def require_inputs(inputs: Sequence[InputMedia], count: int = 1) -> None:
    if len(inputs) < count:
        raise InputError(f"This operation requires at least {count} media input(s)")


def output_path(output_dir: Path, stem: str, extension: str) -> Path:
    clean_extension = extension.lower().lstrip(".")
    if not re.fullmatch(r"[a-z0-9]{2,5}", clean_extension):
        raise InputError("Invalid output format")
    output_dir.mkdir(parents=True, exist_ok=True)
    return output_dir / f"{stem}.{clean_extension}"


def format_param(params: dict[str, Any], default: str, allowed: set[str]) -> str:
    value = str(params.get("format", default)).lower().lstrip(".")
    if value not in allowed:
        raise InputError(
            "Unsupported output format", details={"format": value, "allowed": sorted(allowed)}
        )
    return value


def video_codecs(extension: str, *, crf: int = 23, preset: str = "medium") -> list[str]:
    if extension == "webm":
        return ["-c:v", "libvpx-vp9", "-crf", str(crf), "-b:v", "0", "-c:a", "libopus"]
    if extension == "avi":
        return ["-c:v", "mpeg4", "-q:v", "5", "-c:a", "libmp3lame"]
    if extension in {"mpeg", "ts"}:
        return ["-c:v", "mpeg2video", "-c:a", "mp2"]
    return [
        "-c:v",
        "libx264",
        "-preset",
        preset,
        "-crf",
        str(crf),
        "-c:a",
        "aac",
        "-b:a",
        "128k",
        "-movflags",
        "+faststart",
    ]


def audio_codec(extension: str) -> list[str]:
    return {
        "mp3": ["-c:a", "libmp3lame", "-b:a", "192k"],
        "wav": ["-c:a", "pcm_s16le"],
        "aac": ["-c:a", "aac", "-b:a", "192k"],
        "m4a": ["-c:a", "aac", "-b:a", "192k"],
        "flac": ["-c:a", "flac"],
        "ogg": ["-c:a", "libvorbis", "-q:a", "5"],
        "opus": ["-c:a", "libopus", "-b:a", "128k"],
    }[extension]


async def execute(
    ffmpeg: FFmpegService,
    args: Sequence[str | Path],
    output: Path,
    operation: str,
    metadata: dict[str, Any] | None = None,
) -> OperationResult:
    await ffmpeg.run(args, operation=operation)
    if not output.is_file() or output.stat().st_size == 0:
        raise ProcessingError("The operation did not produce an output file")
    return OperationResult(
        path=output,
        filename=output.name,
        mime_type=mimetypes.guess_type(output.name)[0] or "application/octet-stream",
        metadata={"operation": operation, **(metadata or {})},
    )


def even(value: int) -> int:
    return max(2, value if value % 2 == 0 else value - 1)


def subtitle_filter_path(path: Path) -> str:
    value = str(path.resolve()).replace("\\", "\\\\")
    for character in (":", "'", "[", "]", ","):
        value = value.replace(character, f"\\{character}")
    return value


def atempo_chain(factor: float) -> str:
    values: list[float] = []
    remaining = factor
    while remaining > 2.0:
        values.append(2.0)
        remaining /= 2.0
    while remaining < 0.5:
        values.append(0.5)
        remaining /= 0.5
    values.append(remaining)
    return ",".join(f"atempo={value:.6g}" for value in values)