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 audio_codec, execute, output_path, require_inputs, video_codecs from app.services.ffmpeg_service import FFmpegService def _times(params: dict[str, Any]) -> tuple[float, float | None]: try: start = float(params.get("start", 0)) duration = float(params["duration"]) if params.get("duration") is not None else None if duration is None and params.get("end") is not None: duration = float(params["end"]) - start except (TypeError, ValueError) as exc: raise InputError("start, end, and duration must be numbers") from exc if start < 0 or (duration is not None and duration <= 0): raise InputError("start must be non-negative and duration/end must follow start") return start, duration async def trim_video( ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path ) -> OperationResult: require_inputs(inputs) start, duration = _times(params) output = output_path(output_dir, "trimmed", "mp4") args: list[str | Path] = ["-ss", str(start), "-i", inputs[0].temp_path] if duration is not None: args += ["-t", str(duration)] args += [*video_codecs("mp4"), output] return await execute(ffmpeg, args, output, "video.trim", {"start": start, "duration": duration}) async def trim_audio( ffmpeg: FFmpegService, inputs: Sequence[InputMedia], params: dict[str, Any], output_dir: Path ) -> OperationResult: require_inputs(inputs) start, duration = _times(params) output = output_path(output_dir, "trimmed", "mp3") args: list[str | Path] = ["-ss", str(start), "-i", inputs[0].temp_path] if duration is not None: args += ["-t", str(duration)] args += [*audio_codec("mp3"), output] return await execute(ffmpeg, args, output, "audio.trim", {"start": start, "duration": duration})