app/operations/rotate.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 execute, output_path, require_inputs, video_codecs
from app.services.ffmpeg_service import FFmpegService


async def rotate_video(
    ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
    require_inputs(inputs)
    try:
        angle = int(params.get("angle", 90)) % 360
    except (TypeError, ValueError) as exc:
        raise InputError("'angle' must be an integer") from exc
    filters = {90: "transpose=1", 180: "hflip,vflip", 270: "transpose=2", 0: "null"}
    if angle not in filters:
        raise InputError("Rotation angle must be 0, 90, 180, or 270 degrees")
    output = output_path(output_dir, "rotated", "mp4")
    return await execute(
        ffmpeg,
        [
            "-i",
            inputs[0].temp_path,
            "-vf",
            filters[angle],
            "-metadata:s:v:0",
            "rotate=0",
            *video_codecs("mp4"),
            output,
        ],
        output,
        "video.rotate",
        {"angle": angle},
    )


async def reverse_video(
    ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
    require_inputs(inputs)
    output = output_path(output_dir, "reversed", "mp4")
    has_audio = bool(inputs[0].metadata.get("audio_streams", True))
    args: list[str | Path] = ["-i", inputs[0].temp_path, "-vf", "reverse"]
    if has_audio:
        args += ["-af", "areverse"]
    args += [*video_codecs("mp4"), output]
    return await execute(ffmpeg, args, output, "video.reverse")


async def change_speed(
    ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
    from app.operations.common import atempo_chain

    require_inputs(inputs)
    try:
        factor = float(params.get("factor", 2.0))
    except (TypeError, ValueError) as exc:
        raise InputError("'factor' must be a number") from exc
    if not 0.125 <= factor <= 8:
        raise InputError("Speed factor must be between 0.125 and 8")
    output = output_path(output_dir, "speed", "mp4")
    has_audio = bool(inputs[0].metadata.get("audio_streams", True))
    args: list[str | Path] = ["-i", inputs[0].temp_path, "-vf", f"setpts=PTS/{factor:.8g}"]
    if has_audio:
        args += ["-af", atempo_chain(factor)]
    args += [*video_codecs("mp4"), output]
    return await execute(ffmpeg, args, output, "video.speed", {"factor": factor})


async def slow_motion(
    ffmpeg: FFmpegService,
    inputs: Sequence[InputMedia],
    params: dict[str, Any],
    output_dir: Path,
) -> OperationResult:
    result = await change_speed(
        ffmpeg, inputs, {"factor": params.get("factor", 0.5), **params}, output_dir
    )
    result.metadata["operation"] = "video.slow_motion"
    return result


async def change_fps(
    ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
    require_inputs(inputs)
    try:
        fps = float(params.get("fps", 30))
    except (TypeError, ValueError) as exc:
        raise InputError("'fps' must be a number") from exc
    if not 1 <= fps <= 240:
        raise InputError("FPS must be between 1 and 240")
    output = output_path(output_dir, "fps", "mp4")
    return await execute(
        ffmpeg,
        ["-i", inputs[0].temp_path, "-vf", f"fps={fps:.6g}", *video_codecs("mp4"), output],
        output,
        "video.fps",
        {"fps": fps},
    )


async def change_bitrate(
    ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
    require_inputs(inputs)
    bitrate = str(params.get("bitrate", "1500k"))
    normalized = bitrate.lower().removesuffix("k").removesuffix("m")
    if not normalized.replace(".", "", 1).isdigit():
        raise InputError("Invalid bitrate; examples: 1500k or 2M")
    output = output_path(output_dir, "bitrate", "mp4")
    args = [
        "-i",
        inputs[0].temp_path,
        "-c:v",
        "libx264",
        "-b:v",
        bitrate,
        "-maxrate",
        bitrate,
        "-bufsize",
        bitrate,
        "-c:a",
        "aac",
        "-b:a",
        "128k",
        "-movflags",
        "+faststart",
        output,
    ]
    return await execute(ffmpeg, args, output, "video.bitrate", {"bitrate": bitrate})