Spaces:
Running
Running
File size: 3,064 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 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 | 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 (
VIDEO_FORMATS,
execute,
format_param,
output_path,
require_inputs,
video_codecs,
)
from app.services.ffmpeg_service import FFmpegService
from app.services.validator import bounded_number
async def compress_video(
ffmpeg: FFmpegService,
inputs: Sequence[InputMedia],
params: dict[str, Any],
output_dir: Path,
) -> OperationResult:
require_inputs(inputs)
extension = format_param(params, "mp4", VIDEO_FORMATS)
try:
crf = int(params.get("crf", 28))
except (TypeError, ValueError) as exc:
raise InputError("'crf' must be an integer") from exc
if not 0 <= crf <= 51:
raise InputError("'crf' must be between 0 and 51")
preset = str(params.get("preset", "medium"))
if preset not in {"ultrafast", "superfast", "veryfast", "faster", "fast", "medium", "slow"}:
raise InputError("Unsupported x264 preset")
output = output_path(output_dir, "compressed", extension)
args: list[str | Path] = ["-i", inputs[0].temp_path]
scale = params.get("max_width")
if scale:
try:
max_width = int(scale)
except (TypeError, ValueError) as exc:
raise InputError("'max_width' must be an integer") from exc
args += ["-vf", f"scale='min({max_width},iw)':-2"]
args += video_codecs(extension, crf=crf, preset=preset)
if params.get("bitrate"):
bitrate = str(params["bitrate"])
if not bitrate.replace("k", "").replace("M", "").isdigit():
raise InputError("Invalid video bitrate")
args += ["-b:v", bitrate]
args += [output]
return await execute(ffmpeg, args, output, "video.compress", {"crf": crf, "format": extension})
async def normalize_video(
ffmpeg: FFmpegService,
inputs: Sequence[InputMedia],
params: dict[str, Any],
output_dir: Path,
) -> OperationResult:
require_inputs(inputs)
output = output_path(output_dir, "normalized", "mp4")
args: list[str | Path] = ["-i", inputs[0].temp_path]
if inputs[0].metadata.get("audio_streams"):
args += ["-af", "loudnorm=I=-16:LRA=11:TP=-1.5"]
args += [*video_codecs("mp4", crf=23), output]
return await execute(ffmpeg, args, output, "video.normalize")
async def normalize_audio(
ffmpeg: FFmpegService,
inputs: Sequence[InputMedia],
params: dict[str, Any],
output_dir: Path,
) -> OperationResult:
require_inputs(inputs)
target = bounded_number(params, "target_lufs", -16, -70, -5)
output = output_path(output_dir, "normalized", "wav")
args = [
"-i",
inputs[0].temp_path,
"-af",
f"loudnorm=I={target}:LRA=11:TP=-1.5",
"-c:a",
"pcm_s16le",
output,
]
return await execute(ffmpeg, args, output, "audio.normalize", {"target_lufs": target})
|