from fastapi import UploadFile, HTTPException, status from PIL import Image from io import BytesIO from pathlib import Path import uuid from typing import Optional from PIL import Image from psd_tools import PSDImage from app.core.logging import setup_logger from app.core.config import get_settings from app.core.constants import ( ERR_IMG_WIDTH_HEIGHT, ERR_FILE_TOO_LARGE, ERR_PSD_EXT, ERR_OVERLAY_EXT ) logger = setup_logger(__name__) settings = get_settings() # Limits CHUNK_SIZE = settings.CHUNK_SIZE MAX_PROFILE_IMG_SIZE = settings.MAX_PROFILE_IMG_SIZE MAX_PSD_SIZE = settings.MAX_PRODUCT_IMG_SIZE MAX_PRODUCT_IMG_SIZE = settings.MAX_PRODUCT_IMG_SIZE MAX_UPLOAD_BYTES = settings.MAX_PRODUCT_UPLOAD_BYTES def validate_profile_image(file: UploadFile): image_data = file.file.read() image = Image.open(BytesIO(image_data)) image_format = image.format.lower() allowed_formats = ['jpeg', 'png'] if image_format not in allowed_formats: raise ValueError(f"Invalid image format. Only {', '.join(allowed_formats)} are allowed.") if len(image_data) > MAX_PROFILE_IMG_SIZE: raise ValueError(f"File is too large. Maximum file size is {MAX_PROFILE_IMG_SIZE / 1024 / 1024:.2f}MB.") file.file.seek(0) def validate_product_images(file: UploadFile): image_data = file.file.read() image = Image.open(BytesIO(image_data)) image_format = image.format.lower() allowed_formats = ["psd", "png", "jpg", "jpeg"] if image_format not in allowed_formats: raise ValueError(f"Invalid image format. Only {', '.join(allowed_formats)} are allowed.") if len(image_data) > MAX_PRODUCT_IMG_SIZE: raise ValueError(f"File is too large. Maximum file size is {MAX_PRODUCT_IMG_SIZE / 1024 / 1024:.2f}MB.") file.file.seek(0) def validate_base_file(file: UploadFile): allowed = ["psd"] ext = file.filename.split(".")[-1].lower() if ext not in allowed: raise ValueError(ERR_PSD_EXT) file.file.seek(0) def validate_overlay_file(file: UploadFile): allowed = ["png", "jpg", "jpeg"] ext = file.filename.split(".")[-1].lower() if ext not in allowed: raise ValueError(ERR_OVERLAY_EXT) file.file.seek(0) def validate_logo_file(file: UploadFile): allowed = ["png", "jpg", "jpeg"] ext = file.filename.split(".")[-1].lower() if ext not in allowed: raise ValueError("Invalid logo/stamp image. Must be PNG/JPG only.") file.file.seek(0) def safe_delete(path: Optional[Path]) -> None: """ Safely delete a file from disk. This helper removes files while silently ignoring missing paths and logging any unexpected exceptions. Args: path (Optional[Path]): Path to delete. No action if None. Notes: - Used for cleanup during partial failures. - Never raises exceptions to the caller. """ if not path: return try: if path.exists(): path.unlink() except Exception: logger.exception("Failed to delete file: %s", path) async def save_upload_file_streamed( upload: UploadFile, directory: Path, max_bytes: int = MAX_UPLOAD_BYTES ) -> Path: """ Save an uploaded file to disk using streaming I/O. This prevents loading large files fully into RAM, making it safe for high-volume uploads. Args: upload (UploadFile): File uploaded by the client. directory (Path): Directory where the file should be saved. max_bytes (int): Maximum allowed upload size in bytes. Returns: Path: Full path to the saved file. Raises: HTTPException: 413 if the file exceeds the maximum allowed size. Notes: - Uses CHUNK_SIZE to read file incrementally. - Filename is sanitized and made unique using UUID. """ validate_product_images(upload) # Sanitize user filename filename = Path(upload.filename or "upload").name unique_filename = f"{uuid.uuid4().hex}_{filename}" file_path = directory / unique_filename bytes_written = 0 try: with file_path.open("wb") as buffer: while True: chunk = await upload.read(CHUNK_SIZE) if not chunk: break buffer.write(chunk) bytes_written += len(chunk) if bytes_written > max_bytes: buffer.close() safe_delete(file_path) raise HTTPException(status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, detail=ERR_FILE_TOO_LARGE) finally: try: await upload.close() except Exception: pass return file_path def resize_and_crop_to_fill(img: Image.Image, target_w: int, target_h: int) -> Image.Image: """ Resize an image while maintaining aspect ratio, then center-crop it. Ensures the output completely fills the target size without letterboxing. Args: img (Image.Image): Input PIL image. target_w (int): Output width. target_h (int): Output height. Returns: Image.Image: Resized + cropped PIL image. Raises: ValueError: If image has zero width/height. Notes: - Uses LANCZOS for high-quality resampling. - Commonly used for fitting overlay images into PSD placeholders. """ ow, oh = img.size if ow == 0 or oh == 0: raise ValueError(ERR_IMG_WIDTH_HEIGHT) scale = max(target_w / ow, target_h / oh) new_w, new_h = int(ow * scale), int(oh * scale) img_resized = img.resize((new_w, new_h), Image.Resampling.LANCZOS) left = (new_w - target_w) // 2 upper = (new_h - target_h) // 2 return img_resized.crop((left, upper, left + target_w, upper + target_h)) def render_layer_alpha_fullsize(psd: PSDImage, target_layer, canvas_size): """ Render the full-size alpha mask of a given PSD layer. This isolates a single layer by temporarily hiding all others, composites the PSD, extracts the alpha channel, and restores visibility. Args: psd (PSDImage): Loaded PSD object. target_layer: PSD layer whose alpha mask is required. canvas_size (tuple): Target (width, height) output resolution. Returns: Image.Image: Grayscale ("L") alpha mask for the layer. Notes: - Always returns an image even if rendering fails. - Non-blocking, used heavily during overlay placement. """ try: # Save visibility of all layers vis_state = [ (layer, getattr(layer, "visible", True)) for layer in psd.descendants() ] # Hide all layers for layer, _ in vis_state: try: layer.visible = False except Exception: pass # Show only target layer try: target_layer.visible = True except Exception: pass rendered = psd.composite().convert("RGBA") alpha = rendered.split()[-1] # Restore original visibility for layer, original_vis in vis_state: try: layer.visible = original_vis except Exception: pass # Resize alpha if needed if alpha.size != canvas_size: alpha = alpha.resize(canvas_size, Image.Resampling.LANCZOS) return alpha except Exception: logger.exception("Failed to compute layer alpha mask") return Image.new("L", canvas_size, 255)