Spaces:
Paused
Paused
File size: 4,184 Bytes
0673086 | 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 | """Validate browser image bytes and create request-scoped, private image files."""
import base64
import binascii
import io
import shutil
import tempfile
import warnings
from pathlib import Path
from PIL import Image, ImageOps, UnidentifiedImageError
MAX_UPLOAD_BYTES = 12 * 1024 * 1024
MAX_UPLOAD_TOTAL = 32 * 1024 * 1024
MAX_IMAGES = 8
MAX_IMAGE_PIXELS = 16_000_000
FORMATS = {"image/jpeg": "JPEG", "image/png": "PNG", "image/webp": "WEBP", "image/gif": "GIF"}
def materialize_images(request):
images = request.get("images", [])
if not isinstance(images, list) or len(images) > MAX_IMAGES:
raise ValueError("images must be a list of at most 8 uploaded files.")
if not images:
return request, None
# mkdtemp creates a private directory; client filenames never become paths.
directory = Path(tempfile.mkdtemp(prefix="interndecision-upload-"))
paths, total, normalized_total = [], 0, 0
try:
for index, item in enumerate(images):
if not isinstance(item, dict) or not isinstance(item.get("data"), str):
raise ValueError("Upload image bytes; local paths and remote URLs are not accepted.")
data = item["data"]
if len(data) > 4 * ((MAX_UPLOAD_BYTES + 2) // 3) + 128:
raise ValueError("Each image is limited to 12 MB.")
if data.startswith("data:"):
header, separator, encoded = data.partition(",")
if not separator or not header.endswith(";base64"):
raise ValueError("Images must use base64 data URLs.")
content_type = header[5:].split(";")[0].lower()
else:
encoded, content_type = data, str(item.get("type", "")).lower()
if content_type not in FORMATS:
raise ValueError("Supported image formats: JPEG, PNG, WebP and static GIF.")
try:
payload = base64.b64decode(encoded, validate=True)
except (ValueError, binascii.Error) as exc:
raise ValueError("Invalid image base64.") from exc
if not payload or len(payload) > MAX_UPLOAD_BYTES:
raise ValueError("Each image must be between 1 byte and 12 MB.")
total += len(payload)
if total > MAX_UPLOAD_TOTAL:
raise ValueError("Combined image upload is limited to 32 MB.")
try:
with warnings.catch_warnings():
warnings.simplefilter("error", Image.DecompressionBombWarning)
with Image.open(io.BytesIO(payload)) as source:
if source.format != FORMATS[content_type]:
raise ValueError("Image content does not match its declared format.")
if source.width * source.height > MAX_IMAGE_PIXELS:
raise ValueError("Each image is limited to 16 million pixels.")
if getattr(source, "n_frames", 1) != 1:
raise ValueError("Animated images are not supported; upload a still image.")
source.load()
# Apply camera orientation, strip metadata and preserve pixel resolution.
image = ImageOps.exif_transpose(source).convert("RGB")
image.info.clear()
path = directory / f"{index:02d}.png"
image.save(path, format="PNG")
image.close()
except (
UnidentifiedImageError,
OSError,
Image.DecompressionBombError,
Image.DecompressionBombWarning,
) as exc:
raise ValueError("Invalid, incomplete or oversized image.") from exc
normalized_total += path.stat().st_size
if normalized_total > MAX_UPLOAD_TOTAL:
raise ValueError("Decoded images exceed 32 MB; upload smaller images.")
paths.append(str(path))
return {**request, "images": paths}, directory
except BaseException:
shutil.rmtree(directory, ignore_errors=True)
raise
|