app/operations/extract_audio.py

from __future__ import annotations

import re
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 audio_codec, execute, output_path, require_inputs
from app.services.ffmpeg_service import FFmpegService
from app.services.validator import bounded_number


async def extract_audio(
    ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
    require_inputs(inputs)
    extension = str(params.get("format", "mp3")).lower().lstrip(".")
    if extension not in {"mp3", "wav", "aac", "m4a", "flac", "ogg", "opus"}:
        raise InputError("Unsupported extracted audio format")
    output = output_path(output_dir, "audio", extension)
    return await execute(
        ffmpeg,
        ["-i", inputs[0].temp_path, "-vn", *audio_codec(extension), output],
        output,
        "audio.extract",
        {"format": extension},
    )


async def replace_audio(
    ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
    require_inputs(inputs, 2)
    output = output_path(output_dir, "replaced-audio", "mp4")
    args = [
        "-i",
        inputs[0].temp_path,
        "-i",
        inputs[1].temp_path,
        "-map",
        "0:v:0",
        "-map",
        "1:a:0",
        "-c:v",
        "copy",
        "-c:a",
        "aac",
        "-shortest",
        "-movflags",
        "+faststart",
        output,
    ]
    return await execute(ffmpeg, args, output, "video.replace_audio")


async def remove_audio(
    ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
    require_inputs(inputs)
    output = output_path(output_dir, "muted", "mp4")
    return await execute(
        ffmpeg,
        ["-i", inputs[0].temp_path, "-an", "-c:v", "copy", output],
        output,
        "video.remove_audio",
    )


async def mute_video(
    ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
    result = await remove_audio(ffmpeg, inputs, params, output_dir)
    result.metadata["operation"] = "video.mute"
    return result


async def fade_audio(
    ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
    require_inputs(inputs)
    start = bounded_number(params, "start", 0, 0, 86_400)
    duration = bounded_number(params, "duration", 3, 0.01, 86_400)
    fade_type = str(params.get("type", "in"))
    if fade_type not in {"in", "out"}:
        raise InputError("Fade requires type in/out, non-negative start, and positive duration")
    output = output_path(output_dir, "faded", "mp3")
    filter_name = "afade=t=in" if fade_type == "in" else "afade=t=out"
    return await execute(
        ffmpeg,
        [
            "-i",
            inputs[0].temp_path,
            "-af",
            f"{filter_name}:st={start}:d={duration}",
            *audio_codec("mp3"),
            output,
        ],
        output,
        "audio.fade",
    )


async def set_volume(
    ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
    require_inputs(inputs)
    volume = bounded_number(params, "volume", 1, 0, 10)
    output = output_path(output_dir, "volume", "mp3")
    return await execute(
        ffmpeg,
        ["-i", inputs[0].temp_path, "-af", f"volume={volume}", *audio_codec("mp3"), output],
        output,
        "audio.volume",
        {"volume": volume},
    )


async def remove_silence(
    ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
    require_inputs(inputs)
    threshold = str(params.get("threshold", "-45dB"))
    if not re.fullmatch(r"-?\d+(?:\.\d+)?dB", threshold):
        raise InputError("threshold must be a decibel value such as -45dB")
    output = output_path(output_dir, "no-silence", "mp3")
    return await execute(
        ffmpeg,
        [
            "-i",
            inputs[0].temp_path,
            "-af",
            f"silenceremove=start_periods=1:start_duration=0.2:start_threshold={threshold}:stop_periods=-1:stop_duration=0.2:stop_threshold={threshold}",
            *audio_codec("mp3"),
            output,
        ],
        output,
        "audio.remove_silence",
    )


async def noise_reduction(
    ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path
) -> OperationResult:
    require_inputs(inputs)
    output = output_path(output_dir, "denoised", "mp3")
    return await execute(
        ffmpeg,
        ["-i", inputs[0].temp_path, "-af", "afftdn", *audio_codec("mp3"), output],
        output,
        "audio.noise_reduction",
    )