Spaces:
Running
Running
File size: 3,671 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 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 | from __future__ import annotations
import mimetypes
import re
from collections.abc import Sequence
from pathlib import Path
from typing import Any
from app.core.exceptions import InputError, ProcessingError
from app.models.media import InputMedia, OperationResult
from app.services.ffmpeg_service import FFmpegService
VIDEO_FORMATS = {"mp4", "mkv", "webm", "mov", "avi", "m4v", "mpeg", "ts"}
AUDIO_FORMATS = {"mp3", "wav", "aac", "m4a", "flac", "ogg", "opus"}
IMAGE_FORMATS = {"jpg", "jpeg", "png", "webp", "bmp", "tiff", "gif"}
def require_inputs(inputs: Sequence[InputMedia], count: int = 1) -> None:
if len(inputs) < count:
raise InputError(f"This operation requires at least {count} media input(s)")
def output_path(output_dir: Path, stem: str, extension: str) -> Path:
clean_extension = extension.lower().lstrip(".")
if not re.fullmatch(r"[a-z0-9]{2,5}", clean_extension):
raise InputError("Invalid output format")
output_dir.mkdir(parents=True, exist_ok=True)
return output_dir / f"{stem}.{clean_extension}"
def format_param(params: dict[str, Any], default: str, allowed: set[str]) -> str:
value = str(params.get("format", default)).lower().lstrip(".")
if value not in allowed:
raise InputError(
"Unsupported output format", details={"format": value, "allowed": sorted(allowed)}
)
return value
def video_codecs(extension: str, *, crf: int = 23, preset: str = "medium") -> list[str]:
if extension == "webm":
return ["-c:v", "libvpx-vp9", "-crf", str(crf), "-b:v", "0", "-c:a", "libopus"]
if extension == "avi":
return ["-c:v", "mpeg4", "-q:v", "5", "-c:a", "libmp3lame"]
if extension in {"mpeg", "ts"}:
return ["-c:v", "mpeg2video", "-c:a", "mp2"]
return [
"-c:v",
"libx264",
"-preset",
preset,
"-crf",
str(crf),
"-c:a",
"aac",
"-b:a",
"128k",
"-movflags",
"+faststart",
]
def audio_codec(extension: str) -> list[str]:
return {
"mp3": ["-c:a", "libmp3lame", "-b:a", "192k"],
"wav": ["-c:a", "pcm_s16le"],
"aac": ["-c:a", "aac", "-b:a", "192k"],
"m4a": ["-c:a", "aac", "-b:a", "192k"],
"flac": ["-c:a", "flac"],
"ogg": ["-c:a", "libvorbis", "-q:a", "5"],
"opus": ["-c:a", "libopus", "-b:a", "128k"],
}[extension]
async def execute(
ffmpeg: FFmpegService,
args: Sequence[str | Path],
output: Path,
operation: str,
metadata: dict[str, Any] | None = None,
) -> OperationResult:
await ffmpeg.run(args, operation=operation)
if not output.is_file() or output.stat().st_size == 0:
raise ProcessingError("The operation did not produce an output file")
return OperationResult(
path=output,
filename=output.name,
mime_type=mimetypes.guess_type(output.name)[0] or "application/octet-stream",
metadata={"operation": operation, **(metadata or {})},
)
def even(value: int) -> int:
return max(2, value if value % 2 == 0 else value - 1)
def subtitle_filter_path(path: Path) -> str:
value = str(path.resolve()).replace("\\", "\\\\")
for character in (":", "'", "[", "]", ","):
value = value.replace(character, f"\\{character}")
return value
def atempo_chain(factor: float) -> str:
values: list[float] = []
remaining = factor
while remaining > 2.0:
values.append(2.0)
remaining /= 2.0
while remaining < 0.5:
values.append(0.5)
remaining /= 0.5
values.append(remaining)
return ",".join(f"atempo={value:.6g}" for value in values)
|