Upload 3 files

utils/advanced.py

@@ -0,0 +1,91 @@
+import hashlib
+import numpy as np
+from PIL import Image
+import imagehash
+from skimage.metrics import structural_similarity as ssim
+import cv2
+
+class AdvancedFeatures:
+    @staticmethod
+    def perceptual_hash(image_path):
+        """Generate perceptual hash for deduplication"""
+        img = Image.open(image_path)
+        return imagehash.phash(img, hash_size=16)
+    
+    @staticmethod
+    def find_similar_images(image_paths, threshold=5):
+        """Group similar images by perceptual hash"""
+        hashes = {}
+        for path in image_paths:
+            try:
+                hashes[path] = AdvancedFeatures.perceptual_hash(path)
+            except:
+                pass
+        
+        # Group images
+        groups = []
+        processed = set()
+        
+        for path1, hash1 in hashes.items():
+            if path1 in processed:
+                continue
+            
+            group = [path1]
+            for path2, hash2 in hashes.items():
+                if path2 != path1 and path2 not in processed:
+                    if hash1 - hash2 <= threshold:
+                        group.append(path2)
+                        processed.add(path2)
+            
+            groups.append(group)
+            processed.add(path1)
+        
+        return groups
+    
+    @staticmethod
+    def calculate_ssim(original_path, compressed_path):
+        """Calculate structural similarity index"""
+        img1 = cv2.imread(str(original_path))
+        img2 = cv2.imread(str(compressed_path))
+        
+        # Resize to same dimensions
+        if img1.shape != img2.shape:
+            img2 = cv2.resize(img2, (img1.shape[1], img1.shape[0]))
+        
+        # Convert to grayscale for SSIM
+        gray1 = cv2.cvtColor(img1, cv2.COLOR_BGR2GRAY)
+        gray2 = cv2.cvtColor(img2, cv2.COLOR_BGR2GRAY)
+        
+        score = ssim(gray1, gray2)
+        return score
+    
+    @staticmethod
+    def recursive_optimize(compressor, img, target_size_kb, max_iterations=10):
+        """Find optimal quality to hit target file size"""
+        low, high = 30, 100
+        best_result = None
+        best_size_diff = float('inf')
+        
+        for _ in range(max_iterations):
+            mid = (low + high) // 2
+            
+            # Test at quality 'mid'
+            test_output = io.BytesIO()
+            img.save(test_output, format='WEBP', quality=mid)
+            size_kb = len(test_output.getvalue()) / 1024
+            
+            diff = abs(size_kb - target_size_kb)
+            
+            if diff < best_size_diff:
+                best_size_diff = diff
+                best_result = (mid, test_output.getvalue())
+            
+            if size_kb > target_size_kb:
+                high = mid - 1
+            else:
+                low = mid + 1
+            
+            if diff < target_size_kb * 0.05:  # Within 5%
+                break
+        
+        return best_result

utils/ai_analyzer.py

@@ -0,0 +1,80 @@
+import cv2
+import numpy as np
+import torch
+import torch.nn as nn
+from PIL import Image
+import torchvision.transforms as transforms
+
+class ContentAnalyzer:
+    def __init__(self):
+        # Use lightweight model (MobileNetV2) - works on CPU
+        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
+        self.model = self._load_model()
+        
+    def _load_model(self):
+        # Load pre-trained MobileNetV2 for feature extraction
+        model = torch.hub.load('pytorch/vision:v0.10.0', 'mobilenet_v2', pretrained=True)
+        # Remove classification head to get features
+        model.classifier = nn.Identity()
+        model.eval()
+        return model.to(self.device)
+    
+    def analyze(self, image_path):
+        """Returns image type and optimal compression strategy"""
+        # Load image
+        img = cv2.imread(str(image_path))
+        if img is None:
+            return "photo", {"method": "avif", "quality": 75}
+        
+        h, w = img.shape[:2]
+        
+        # 1. Edge detection (for text/screenshots)
+        gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
+        edges = cv2.Canny(gray, 50, 150)
+        edge_ratio = np.sum(edges > 0) / edges.size
+        
+        # 2. Color analysis
+        unique_colors = len(np.unique(img.reshape(-1, img.shape[2]), axis=0))
+        
+        # 3. Texture analysis (variance of Laplacian)
+        laplacian_var = cv2.Laplacian(gray, cv2.CV_64F).var()
+        
+        # 4. Check for transparency
+        has_transparency = False
+        try:
+            pil_img = Image.open(image_path)
+            has_transparency = pil_img.mode in ('RGBA', 'LA', 'P') and 'transparency' in pil_img.info
+        except:
+            pass
+        
+        # Decision logic
+        if has_transparency:
+            return "graphic_with_transparency", {
+                "method": "png_optimized",
+                "colors": 256,
+                "lossless": True
+            }
+        elif edge_ratio > 0.15 and unique_colors < 5000:
+            return "screenshot_or_text", {
+                "method": "webp_lossless",
+                "quality": 90,
+                "preserve_text": True
+            }
+        elif unique_colors < 1000:
+            return "graphic", {
+                "method": "png_quantized",
+                "colors": min(256, unique_colors),
+                "dither": 0.5
+            }
+        elif laplacian_var < 100:
+            return "smooth_gradient", {
+                "method": "avif",
+                "quality": 80,
+                "avoid_banding": True
+            }
+        else:
+            return "photo", {
+                "method": "avif",
+                "quality": 75,
+                "chroma_subsampling": "4:2:0"
+            }

utils/compressor.py

@@ -0,0 +1,132 @@
+import io
+import subprocess
+from PIL import Image, ImageOps
+import cv2
+import numpy as np
+from pathlib import Path
+import piexif
+
+class ImageCompressor:
+    def __init__(self):
+        self.supported_formats = {'.jpg', '.jpeg', '.png', '.webp', '.gif', '.bmp', '.tiff'}
+        
+    def compress(self, input_path, output_path, method="auto", **kwargs):
+        """Main compression entry point"""
+        img = Image.open(input_path)
+        original_size = Path(input_path).stat().st_size
+        
+        # Get original format
+        original_format = img.format
+        
+        if method == "auto":
+            # Use AI analyzer to determine best method
+            from .ai_analyzer import ContentAnalyzer
+            analyzer = ContentAnalyzer()
+            _, strategy = analyzer.analyze(input_path)
+            method = strategy.get("method", "avif")
+            kwargs = {**kwargs, **strategy}
+        
+        # Apply compression based on method
+        if method == "avif":
+            compressed = self._compress_avif(img, kwargs.get('quality', 75))
+            output_path = str(output_path).replace(Path(output_path).suffix, '.avif')
+        elif method == "webp_lossless":
+            compressed = self._compress_webp_lossless(img, kwargs.get('quality', 90))
+            output_path = str(output_path).replace(Path(output_path).suffix, '.webp')
+        elif method == "png_optimized":
+            compressed = self._compress_png_optimized(img, kwargs.get('colors', 256))
+            output_path = str(output_path).replace(Path(output_path).suffix, '.png')
+        elif method == "png_quantized":
+            compressed = self._compress_png_quantized(img, kwargs.get('colors', 128))
+            output_path = str(output_path).replace(Path(output_path).suffix, '.png')
+        elif method == "jpeg_high_quality":
+            compressed = self._compress_jpeg(img, kwargs.get('quality', 85))
+            output_path = str(output_path).replace(Path(output_path).suffix, '.jpg')
+        else:
+            # Default to WebP
+            compressed = self._compress_webp(img, kwargs.get('quality', 80))
+            output_path = str(output_path).replace(Path(output_path).suffix, '.webp')
+        
+        # Save compressed image
+        with open(output_path, 'wb') as f:
+            f.write(compressed)
+        
+        compressed_size = Path(output_path).stat().st_size
+        ratio = (1 - compressed_size / original_size) * 100
+        
+        return output_path, ratio
+    
+    def _compress_avif(self, img, quality=75):
+        """AVIF compression - best for photos"""
+        output = io.BytesIO()
+        
+        # Convert RGBA to RGB if needed (AVIF doesn't support alpha well)
+        if img.mode == 'RGBA':
+            background = Image.new('RGB', img.size, (255, 255, 255))
+            background.paste(img, mask=img.split()[-1])
+            img = background
+        
+        # Save as AVIF (requires pillow-avif-plugin)
+        img.save(output, format='AVIF', quality=quality, speed=4)
+        return output.getvalue()
+    
+    def _compress_webp_lossless(self, img, quality=90):
+        """Lossless WebP for text/screenshots"""
+        output = io.BytesIO()
+        img.save(output, format='WEBP', lossless=True, quality=quality, method=6)
+        return output.getvalue()
+    
+    def _compress_webp(self, img, quality=80):
+        """Standard WebP compression"""
+        output = io.BytesIO()
+        img.save(output, format='WEBP', quality=quality, method=4)
+        return output.getvalue()
+    
+    def _compress_png_optimized(self, img, colors=256):
+        """PNG with color palette reduction"""
+        output = io.BytesIO()
+        
+        # Convert to palette mode
+        if colors < 256:
+            img = img.quantize(colors=colors, method=Image.MEDIANCUT)
+            img.save(output, format='PNG', optimize=True)
+        else:
+            img.save(output, format='PNG', optimize=True)
+        
+        return output.getvalue()
+    
+    def _compress_png_quantized(self, img, colors=128):
+        """Heavy PNG quantization for graphics"""
+        output = io.BytesIO()
+        
+        # Reduce colors dramatically
+        img_quantized = img.quantize(colors=colors, method=Image.FASTOCTREE)
+        img_quantized.save(output, format='PNG', optimize=True)
+        
+        return output.getvalue()
+    
+    def _compress_jpeg(self, img, quality=85):
+        """JPEG compression"""
+        output = io.BytesIO()
+        
+        # Convert to RGB if needed
+        if img.mode in ('RGBA', 'LA', 'P'):
+            img = img.convert('RGB')
+        
+        img.save(output, format='JPEG', quality=quality, optimize=True, progressive=True)
+        return output.getvalue()
+    
+    def smart_resize(self, img, target_dimension=2048):
+        """Intelligently resize if image is too large"""
+        width, height = img.size
+        max_dim = max(width, height)
+        
+        if max_dim > target_dimension:
+            scale = target_dimension / max_dim
+            new_size = (int(width * scale), int(height * scale))
+            return img.resize(new_size, Image.Resampling.LANCZOS)
+        return img
+
+    def remove_metadata(self, image_path):
+        """Strip all EXIF metadata"""
+        piexif.remove(str(image_path))