Spaces:
Sleeping
Sleeping
File size: 4,574 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 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 | from __future__ import annotations
import mimetypes
import re
from pathlib import Path
from typing import Any
from app.core.config import Settings
from app.core.exceptions import InputError
MEDIA_EXTENSIONS = {
".3gp",
".aac",
".aiff",
".apng",
".avi",
".bin",
".bmp",
".flac",
".gif",
".heic",
".jpeg",
".jpg",
".m4a",
".m4v",
".mkv",
".m3u",
".m3u8",
".mov",
".mp3",
".mp4",
".mpeg",
".mpg",
".oga",
".ogg",
".opus",
".png",
".svg",
".tif",
".tiff",
".ts",
".vtt",
".wav",
".webm",
".webp",
".wmv",
".srt",
".ass",
".ssa",
}
class MediaValidator:
def __init__(self, settings: Settings) -> None:
self.settings = settings
def safe_filename(self, filename: str | None, fallback: str = "input.bin") -> str:
name = Path(filename or fallback).name.replace("\x00", "")
if not name or name in {".", ".."}:
raise InputError("Invalid filename")
return name[:240]
def validate_declared(self, filename: str, mime_type: str, size: int) -> None:
if size <= 0:
raise InputError("The uploaded media is empty")
if size > self.settings.max_upload_size:
raise InputError(
"The media exceeds MAX_UPLOAD_SIZE",
details={"size": size, "maximum": self.settings.max_upload_size},
)
suffix = Path(filename).suffix.lower()
if suffix and suffix not in MEDIA_EXTENSIONS:
raise InputError("Unsupported media file extension", details={"extension": suffix})
normalized_mime = mime_type.split(";", 1)[0].lower()
allowed = normalized_mime.startswith(
("audio/", "video/", "image/", "text/")
) or normalized_mime in {
"application/octet-stream",
"application/x-subrip",
"application/vnd.apple.mpegurl",
"application/x-mpegurl",
}
if normalized_mime and not allowed:
raise InputError("Unsupported media MIME type", details={"mime_type": mime_type})
def validate_probe(self, metadata: dict[str, Any]) -> None:
streams = [
*(metadata.get("video_streams") or []),
*(metadata.get("audio_streams") or []),
]
if not streams:
raise InputError("The input has no decodable video or audio stream")
for stream in streams:
codec = stream.get("codec")
if not isinstance(codec, str) or not re.fullmatch(r"[a-zA-Z0-9_.-]{1,64}", codec):
raise InputError("The input contains an invalid or unsupported codec")
duration = metadata.get("duration")
if duration and float(duration) > self.settings.max_duration_seconds:
raise InputError(
"Media duration exceeds the configured limit",
details={"duration": duration, "maximum": self.settings.max_duration_seconds},
)
resolution = metadata.get("resolution") or {}
width, height = resolution.get("width", 0), resolution.get("height", 0)
if width and height and width * height > self.settings.max_resolution_pixels:
raise InputError(
"Media resolution exceeds the configured limit",
details={"width": width, "height": height},
)
@staticmethod
def infer_mime(path: Path, fallback: str = "application/octet-stream") -> str:
return mimetypes.guess_type(path.name)[0] or fallback
def as_bool(value: Any, default: bool = False) -> bool:
if value is None:
return default
if isinstance(value, bool):
return value
return str(value).strip().lower() in {"1", "true", "yes", "on"}
def bounded_number(
params: dict[str, Any], name: str, default: float, minimum: float, maximum: float
) -> float:
try:
value = float(params.get(name, default))
except (TypeError, ValueError) as exc:
raise InputError(f"'{name}' must be a number") from exc
if not minimum <= value <= maximum:
raise InputError(f"'{name}' must be between {minimum} and {maximum}")
return value
def positive_int(params: dict[str, Any], name: str, default: int) -> int:
try:
value = int(params.get(name, default))
except (TypeError, ValueError) as exc:
raise InputError(f"'{name}' must be an integer") from exc
if value <= 0:
raise InputError(f"'{name}' must be greater than zero")
return value
|