app/operations/concat.py

from __future__ import annotations

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

import aiofiles

from app.models.media import InputMedia, OperationResult
from app.operations.common import even, execute, output_path, require_inputs
from app.operations.merge import merge_audio, merge_video
from app.services.ffmpeg_service import FFmpegService
from app.services.validator import as_bool, bounded_number, positive_int


async def concat_video(
    ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
    require_inputs(inputs, 2)
    if not as_bool(params.get("stream_copy"), False):
        return await merge_video(
            ffmpeg, inputs, params, output_dir, operation="video.concat", stem="concatenated"
        )
    manifest = output_dir / "concat.txt"
    async with aiofiles.open(manifest, "w", encoding="utf-8") as output_file:
        for media in inputs:
            escaped = str(media.temp_path.resolve()).replace("'", "'\\''")
            await output_file.write(f"file '{escaped}'\n")
    output = output_path(output_dir, "concatenated", "mp4")
    return await execute(
        ffmpeg,
        [
            "-f",
            "concat",
            "-safe",
            "0",
            "-i",
            manifest,
            "-c",
            "copy",
            "-movflags",
            "+faststart",
            output,
        ],
        output,
        "video.concat",
        {"inputs": len(inputs), "stream_copy": True},
    )


async def concat_audio(
    ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
    return await merge_audio(
        ffmpeg, inputs, params, output_dir, operation="audio.concat", stem="concatenated"
    )


async def image_slideshow(
    ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
    require_inputs(inputs)
    duration = bounded_number(params, "duration_per_image", 3.0, 0.1, 300)
    width = even(positive_int(params, "width", 1280))
    height = even(positive_int(params, "height", 720))
    fps = positive_int(params, "fps", 30)
    command: list[str | Path] = []
    for media in inputs:
        command += ["-loop", "1", "-t", str(duration), "-i", media.temp_path]
    filters = [
        f"[{index}:v]scale={width}:{height}:force_original_aspect_ratio=decrease,"
        f"pad={width}:{height}:(ow-iw)/2:(oh-ih)/2,setsar=1,format=yuv420p[v{index}]"
        for index in range(len(inputs))
    ]
    links = "".join(f"[v{index}]" for index in range(len(inputs)))
    filters.append(f"{links}concat=n={len(inputs)}:v=1:a=0[v]")
    output = output_path(output_dir, "slideshow", "mp4")
    command += [
        "-filter_complex",
        ";".join(filters),
        "-map",
        "[v]",
        "-r",
        str(fps),
        "-c:v",
        "libx264",
        "-pix_fmt",
        "yuv420p",
        "-movflags",
        "+faststart",
        output,
    ]
    return await execute(
        ffmpeg,
        command,
        output,
        "image.slideshow",
        {"images": len(inputs), "duration_per_image": duration},
    )


async def image_sequence(
    ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
    fps = positive_int(params, "fps", 25)
    result = await image_slideshow(
        ffmpeg,
        inputs,
        {
            "duration_per_image": params.get("duration_per_image", 1 / fps),
            **params,
        },
        output_dir,
    )
    result.metadata["operation"] = "image.sequence"
    return result


async def image_to_video(
    ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
    require_inputs(inputs)
    duration = bounded_number(params, "duration", 5, 0.1, 3600)
    fps = positive_int(params, "fps", 30)
    output = output_path(output_dir, "image-video", "mp4")
    args = [
        "-loop",
        "1",
        "-i",
        inputs[0].temp_path,
        "-t",
        str(duration),
        "-vf",
        "scale=trunc(iw/2)*2:trunc(ih/2)*2,format=yuv420p",
        "-r",
        str(fps),
        "-c:v",
        "libx264",
        "-movflags",
        "+faststart",
        output,
    ]
    return await execute(ffmpeg, args, output, "image.video", {"duration": duration})