Spaces:
Sleeping
Sleeping
| from fastapi import APIRouter, UploadFile, File, Form, HTTPException | |
| from fastapi.responses import StreamingResponse | |
| from PIL import Image | |
| from io import BytesIO | |
| # ========================== | |
| # πΌοΈ Image Processing Router | |
| # ========================== | |
| router = APIRouter( | |
| prefix="/image", | |
| tags=["Image Processing"] | |
| ) | |
| # ========================== | |
| # π¦ JPG Target Size Endpoint | |
| # ========================== | |
| async def compress_jpg_to_size( | |
| file: UploadFile = File(..., description="The JPG image file to compress."), | |
| target_size_kb: int = Form(..., description="Target file size in KB (e.g., 240).") | |
| ): | |
| """ | |
| π§ Compresses a JPEG image to fit within a specific file size (in KB). | |
| Uses binary search to find the best quality and, if necessary, resizes dimensions. | |
| """ | |
| # -------------------------- | |
| # π Validation Checks | |
| # -------------------------- | |
| if file.content_type not in ["image/jpeg", "image/jpg"]: | |
| raise HTTPException( | |
| status_code=400, | |
| detail="Invalid file type. Only JPEG files are supported." | |
| ) | |
| if target_size_kb <= 0: | |
| raise HTTPException( | |
| status_code=400, | |
| detail="Target size must be greater than 0 KB." | |
| ) | |
| # Calculate target bytes | |
| target_bytes = target_size_kb * 1024 | |
| # -------------------------- | |
| # π₯ Read Image Bytes | |
| # -------------------------- | |
| content = await file.read() | |
| # β‘ Optimization: If original is already smaller than target, return original | |
| if len(content) <= target_bytes: | |
| return StreamingResponse( | |
| BytesIO(content), | |
| media_type="image/jpeg", | |
| headers={"Content-Disposition": f"attachment; filename=original_{file.filename}"} | |
| ) | |
| input_buffer = BytesIO(content) | |
| try: | |
| img = Image.open(input_buffer) | |
| # Convert to RGB if necessary (e.g., if input was RGBA) | |
| if img.mode != 'RGB': | |
| img = img.convert('RGB') | |
| output_buffer = BytesIO() | |
| # -------------------------- | |
| # π Binary Search Algorithm | |
| # -------------------------- | |
| # We search for the best quality between 1 and 95 | |
| min_quality = 1 | |
| max_quality = 95 | |
| best_buffer = None | |
| while min_quality <= max_quality: | |
| quality = (min_quality + max_quality) // 2 | |
| # Clear buffer and save with current quality | |
| output_buffer.seek(0) | |
| output_buffer.truncate() | |
| img.save(output_buffer, format="JPEG", quality=quality) | |
| size = output_buffer.tell() | |
| if size <= target_bytes: | |
| # This fits! But can we get better quality? | |
| best_buffer = BytesIO(output_buffer.getvalue()) # Store success | |
| min_quality = quality + 1 # Try higher quality | |
| else: | |
| # Too big, reduce quality | |
| max_quality = quality - 1 | |
| # -------------------------- | |
| # β οΈ Fallback: Resize | |
| # -------------------------- | |
| # If even quality=1 is too big, we must reduce image dimensions | |
| if best_buffer is None: | |
| resize_factor = 0.9 | |
| while True: | |
| width, height = img.size | |
| new_width = int(width * resize_factor) | |
| new_height = int(height * resize_factor) | |
| # Stop if image gets too tiny (sanity check) | |
| if new_width < 10 or new_height < 10: | |
| break | |
| img = img.resize((new_width, new_height), Image.Resampling.LANCZOS) | |
| output_buffer.seek(0) | |
| output_buffer.truncate() | |
| img.save(output_buffer, format="JPEG", quality=5) # Low quality + resize | |
| if output_buffer.tell() <= target_bytes: | |
| best_buffer = output_buffer | |
| break | |
| resize_factor *= 0.9 # Reduce size by another 10% | |
| # If we still failed (extremely rare), just return the last attempt | |
| if best_buffer is None: | |
| output_buffer.seek(0) | |
| best_buffer = output_buffer | |
| best_buffer.seek(0) | |
| # -------------------------- | |
| # π€ Return as Stream | |
| # -------------------------- | |
| return StreamingResponse( | |
| best_buffer, | |
| media_type="image/jpeg", | |
| headers={ | |
| "Content-Disposition": f"attachment; filename=compressed_{file.filename}" | |
| } | |
| ) | |
| except Exception as e: | |
| raise HTTPException( | |
| status_code=500, | |
| detail=f"Image processing error: {e}" | |
| ) |