77 lines
2.3 KiB
Python
77 lines
2.3 KiB
Python
"""Image pipeline. Validates, re-encodes (drops EXIF), thumbnails, saves to media.
|
|
|
|
Re-encoding through Pillow strips metadata and neutralizes polyglot/malicious
|
|
payloads. Filenames are randomized. Returns a ListingImage (uncommitted).
|
|
"""
|
|
import io
|
|
import os
|
|
import secrets
|
|
from PIL import Image
|
|
from flask import current_app
|
|
from app.models.listing import ListingImage
|
|
|
|
MAX_BYTES = 8 * 1024 * 1024 # 8 MB per upload
|
|
MAX_DIM = 1600 # longest edge for the full image
|
|
THUMB_DIM = 400
|
|
ALLOWED = {"JPEG", "PNG", "WEBP"}
|
|
|
|
|
|
class ImageError(ValueError):
|
|
pass
|
|
|
|
|
|
def _media_root():
|
|
root = current_app.config.get("MEDIA_ROOT") or os.path.join(
|
|
current_app.instance_path, "media")
|
|
os.makedirs(root, exist_ok=True)
|
|
return root
|
|
|
|
|
|
def process_upload(file_storage, listing_id, sort_order=0):
|
|
data = file_storage.read()
|
|
if not data:
|
|
raise ImageError("empty file")
|
|
if len(data) > MAX_BYTES:
|
|
raise ImageError("file too large")
|
|
|
|
try:
|
|
img = Image.open(io.BytesIO(data))
|
|
img.verify() # detect truncated/corrupt
|
|
img = Image.open(io.BytesIO(data)) # re-open after verify
|
|
except Exception:
|
|
raise ImageError("not a valid image")
|
|
|
|
if img.format not in ALLOWED:
|
|
raise ImageError("unsupported format")
|
|
|
|
img = img.convert("RGB") # normalize, drop alpha/EXIF
|
|
img.thumbnail((MAX_DIM, MAX_DIM))
|
|
|
|
root = _media_root()
|
|
sub = os.path.join(root, str(listing_id))
|
|
os.makedirs(sub, exist_ok=True)
|
|
name = secrets.token_hex(16)
|
|
|
|
full_rel = os.path.join(str(listing_id), f"{name}.jpg")
|
|
thumb_rel = os.path.join(str(listing_id), f"{name}_t.jpg")
|
|
img.save(os.path.join(root, full_rel), "JPEG", quality=85, optimize=True)
|
|
|
|
thumb = img.copy()
|
|
thumb.thumbnail((THUMB_DIM, THUMB_DIM))
|
|
thumb.save(os.path.join(root, thumb_rel), "JPEG", quality=80, optimize=True)
|
|
|
|
return ListingImage(
|
|
listing_id=listing_id, path=full_rel, thumb_path=thumb_rel,
|
|
sort_order=sort_order, width=img.width, height=img.height,
|
|
)
|
|
|
|
|
|
def delete_image_files(image):
|
|
root = _media_root()
|
|
for rel in (image.path, image.thumb_path):
|
|
if rel:
|
|
try:
|
|
os.remove(os.path.join(root, rel))
|
|
except OSError:
|
|
pass
|