Spaces:
Running
Running
File size: 2,104 Bytes
fba6023 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 | 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})
|