74
lambda/image_processor.py
Normal file
74
lambda/image_processor.py
Normal file
@@ -0,0 +1,74 @@
|
||||
"""Image processing operations"""
|
||||
import io
|
||||
import hashlib
|
||||
from datetime import datetime
|
||||
from PIL import Image
|
||||
|
||||
from config import (
|
||||
ALLOWED_FORMATS, MAX_DIMENSION, MAX_FILE_SIZE,
|
||||
RESIZE_TARGET, THUMBNAIL_TARGET
|
||||
)
|
||||
|
||||
|
||||
def validate_image(img_data: bytes) -> tuple[Image.Image, str]:
|
||||
"""Validate and open image data"""
|
||||
if len(img_data) > MAX_FILE_SIZE:
|
||||
raise ValueError(f'File too large: {len(img_data)} bytes')
|
||||
|
||||
img_hash = hashlib.sha256(img_data).hexdigest()
|
||||
img = Image.open(io.BytesIO(img_data))
|
||||
|
||||
if img.format not in ALLOWED_FORMATS:
|
||||
raise ValueError(f'Invalid format: {img.format}')
|
||||
|
||||
if img.width * img.height > MAX_DIMENSION ** 2:
|
||||
raise ValueError(f'Image too large: {img.width}x{img.height}')
|
||||
|
||||
return img, img_hash
|
||||
|
||||
|
||||
def determine_processing(filename: str) -> tuple[tuple[int, int], str]:
|
||||
"""Determine processing type based on filename"""
|
||||
fn = filename.lower()
|
||||
if '_thumb' in fn:
|
||||
return THUMBNAIL_TARGET, 'resize_200x200'
|
||||
elif '_grayscale' in fn:
|
||||
return None, 'grayscale'
|
||||
else:
|
||||
return RESIZE_TARGET, 'resize_1024x1024'
|
||||
|
||||
|
||||
def process_image(img: Image.Image, target: tuple[int, int], ptype: str) -> Image.Image:
|
||||
"""Apply image processing transformations"""
|
||||
if ptype == 'grayscale':
|
||||
return img.convert('L')
|
||||
elif target:
|
||||
img.thumbnail(target, Image.Resampling.LANCZOS)
|
||||
return img
|
||||
|
||||
|
||||
def save_image(img: Image.Image, original_format: str) -> tuple[bytes, str]:
|
||||
"""Save processed image to bytes"""
|
||||
output = io.BytesIO()
|
||||
fmt = original_format or 'JPEG'
|
||||
img.save(output, format=fmt, quality=85)
|
||||
return output.getvalue(), f'image/{fmt.lower()}'
|
||||
|
||||
|
||||
def get_processed_key(original_key: str) -> str:
|
||||
"""Generate processed object key"""
|
||||
return original_key.replace('uploads/', 'processed/')
|
||||
|
||||
|
||||
def build_result(original_key: str, processed_key: str, orig_size: tuple,
|
||||
img: Image.Image, ptype: str, img_hash: str) -> dict:
|
||||
"""Build processing result dictionary"""
|
||||
return {
|
||||
'status': 'success',
|
||||
'original_size': orig_size,
|
||||
'processed_size': img.size,
|
||||
'processing_type': ptype,
|
||||
'processed_key': processed_key,
|
||||
'timestamp': datetime.utcnow().isoformat(),
|
||||
'hash': img_hash
|
||||
}
|
||||
Reference in New Issue
Block a user