""" Image processing service for company logos and images. """ import logging from pathlib import Path from typing import Optional, Tuple from urllib.parse import urlparse import hashlib import httpx from PIL import Image import io from ..config import get_settings logger = logging.getLogger(__name__) class ImageService: """Service for downloading and processing images.""" # Standard sizes for company images SIZES = { "thumbnail": (150, 150), "medium": (300, 300), "large": (600, 600), } def __init__(self): """Initialize image service.""" settings = get_settings() self.images_dir = settings.images_dir self.images_dir.mkdir(parents=True, exist_ok=True) def _get_image_hash(self, url: str) -> str: """Generate a short hash for an image URL.""" return hashlib.md5(url.encode()).hexdigest()[:12] def _get_image_path( self, company_slug: str, image_type: str = "logo", size: str = "medium" ) -> Path: """ Get the path for storing a company image. Args: company_slug: Company slug for filename image_type: Type of image (logo, photo) size: Size variant (thumbnail, medium, large) Returns: Path to image file """ filename = f"{company_slug}_{image_type}_{size}.webp" return self.images_dir / filename def download_image(self, url: str) -> Optional[bytes]: """ Download an image from URL. Args: url: Image URL Returns: Image bytes or None if download failed """ try: with httpx.Client(timeout=30.0, follow_redirects=True) as client: response = client.get(url) response.raise_for_status() content_type = response.headers.get("content-type", "") if not content_type.startswith("image/"): logger.warning(f"URL is not an image: {url}") return None return response.content except Exception as e: logger.error(f"Failed to download image from {url}: {e}") return None def process_image( self, image_data: bytes, size: Tuple[int, int], quality: int = 85, ) -> Optional[bytes]: """ Process and convert an image to WebP format. Args: image_data: Raw image bytes size: Target size (width, height) quality: WebP quality (1-100) Returns: Processed image bytes or None if processing failed """ try: # Open image img = Image.open(io.BytesIO(image_data)) # Convert to RGB if necessary (for WebP) if img.mode in ("RGBA", "P"): # Create white background for transparency background = Image.new("RGB", img.size, (255, 255, 255)) if img.mode == "P": img = img.convert("RGBA") background.paste(img, mask=img.split()[-1]) img = background elif img.mode != "RGB": img = img.convert("RGB") # Resize maintaining aspect ratio img.thumbnail(size, Image.Resampling.LANCZOS) # Save to WebP output = io.BytesIO() img.save(output, format="WEBP", quality=quality, method=6) return output.getvalue() except Exception as e: logger.error(f"Failed to process image: {e}") return None def save_company_logo( self, company_slug: str, logo_url: str, sizes: Optional[list] = None, ) -> dict: """ Download and save company logo in multiple sizes. Args: company_slug: Company slug for filename logo_url: URL of the logo image sizes: List of size names to generate (default: all) Returns: Dictionary mapping size names to file paths """ if sizes is None: sizes = list(self.SIZES.keys()) saved_paths = {} # Download original image_data = self.download_image(logo_url) if not image_data: return saved_paths # Process and save each size for size_name in sizes: if size_name not in self.SIZES: continue size = self.SIZES[size_name] processed = self.process_image(image_data, size) if processed: path = self._get_image_path(company_slug, "logo", size_name) path.write_bytes(processed) saved_paths[size_name] = str(path) logger.debug(f"Saved {size_name} logo: {path}") return saved_paths def get_logo_urls(self, company_slug: str, base_url: str = "") -> dict: """ Get URLs for company logo images. Args: company_slug: Company slug base_url: Base URL for the images (e.g., /static/images/companies/) Returns: Dictionary mapping size names to URLs """ urls = {} for size_name in self.SIZES: path = self._get_image_path(company_slug, "logo", size_name) if path.exists(): relative_path = path.relative_to(self.images_dir) urls[size_name] = f"{base_url}{relative_path}" return urls def cleanup_company_images(self, company_slug: str) -> int: """ Remove all images for a company. Args: company_slug: Company slug Returns: Number of files removed """ removed = 0 for file in self.images_dir.glob(f"{company_slug}_*"): file.unlink() removed += 1 return removed def get_storage_stats(self) -> dict: """ Get storage statistics for the images directory. Returns: Dictionary with stats """ total_size = 0 file_count = 0 for file in self.images_dir.glob("*.webp"): total_size += file.stat().st_size file_count += 1 return { "file_count": file_count, "total_size_mb": round(total_size / (1024 * 1024), 2), "directory": str(self.images_dir), }