Add dedicated voiceover mixing operation

app/operations/add_voiceover.py

@@ -0,0 +1,118 @@
+from __future__ import annotations
+
+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
+from app.operations.mix_audio import _audio_bitrate, _video_duration
+from app.services.ffmpeg_service import FFmpegService
+from app.services.validator import bounded_number
+
+
+def _bool_param(params: dict[str, Any], name: str, default: bool) -> bool:
+    value = params.get(name, default)
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, (int, float)) and value in (0, 1):
+        return bool(value)
+    if isinstance(value, str):
+        normalized = value.strip().lower()
+        if normalized in {"true", "1", "yes", "on"}:
+            return True
+        if normalized in {"false", "0", "no", "off"}:
+            return False
+    raise InputError(f"{name} must be a boolean")
+
+
+async def add_voiceover(
+    ffmpeg: FFmpegService,
+    inputs: list[InputMedia] | tuple[InputMedia, ...],
+    params: dict[str, Any],
+    output_dir: Path,
+) -> OperationResult:
+    require_inputs(inputs, 2)
+    if len(inputs) != 2:
+        raise InputError("add-voiceover accepts exactly one video and one voiceover audio input")
+
+    duration = _video_duration(inputs)
+    start = bounded_number(params, "start", 0, 0, 86_400)
+    volume = bounded_number(params, "volume", 1, 0, 10)
+    duck_existing = _bool_param(params, "duck_existing", False)
+    duck_volume = bounded_number(params, "duck_volume", 0.35, 0, 1)
+    fade_in = bounded_number(params, "fade_in", 0, 0, 60)
+    fade_out = bounded_number(params, "fade_out", 0, 0, 60)
+    bitrate = _audio_bitrate(params)
+
+    if fade_in + fade_out > duration:
+        raise InputError("fade_in plus fade_out cannot exceed the video duration")
+
+    duration_value = f"{duration:.6f}"
+    delay_ms = round(start * 1000)
+
+    filters: list[str] = [
+        f"[0:a]apad=whole_dur={duration_value},atrim=duration={duration_value}[base]"
+    ]
+
+    voice_filters = [
+        f"[1:a]volume={volume:.6g}",
+        f"adelay={delay_ms}:all=1",
+    ]
+    if fade_in > 0:
+        voice_filters.append(f"afade=t=in:st={start:.6f}:d={fade_in:.6f}")
+    if fade_out > 0:
+        fade_start = max(start, duration - fade_out)
+        voice_filters.append(f"afade=t=out:st={fade_start:.6f}:d={fade_out:.6f}")
+    voice_filters.extend([
+        f"apad=whole_dur={duration_value}",
+        f"atrim=duration={duration_value}[voiceover]",
+    ])
+    filters.append(",".join(voice_filters))
+
+    base_label = "[base]"
+    if duck_existing:
+        # Duck the original mix while the voiceover is active using the
+        # voiceover signal as a sidechain control signal.
+        filters.append(
+            f"[base][voiceover]sidechaincompress=threshold=0.03:ratio=8:attack=20:release=300:"
+            f"makeup=1:mix=1[ducked]"
+        )
+        base_label = "[ducked]"
+
+    filters.append(
+        f"{base_label}[voiceover]amix=inputs=2:duration=first:dropout_transition=0:normalize=0[mixed]"
+    )
+
+    output = output_path(output_dir, "voiceover", "mp4")
+    args: list[str | Path] = []
+    for media in inputs:
+        args.extend(["-i", media.temp_path])
+    args.extend([
+        "-filter_complex", ";".join(filters),
+        "-map", "0:v:0",
+        "-map", "[mixed]",
+        "-c:v", "copy",
+        "-c:a", "aac",
+        "-b:a", bitrate,
+        "-movflags", "+faststart",
+        output,
+    ])
+
+    return await execute(
+        ffmpeg,
+        args,
+        output,
+        "video.add_voiceover",
+        {
+            "start": start,
+            "volume": volume,
+            "fade_in": fade_in,
+            "fade_out": fade_out,
+            "duck_existing": duck_existing,
+            "duck_volume": duck_volume,
+            "preserve_video": True,
+            "mix_mode": "voiceover",
+            "duration_seconds": duration,
+        },
+    )