"""Image save / resize / path management utilities.""" import io import logging import uuid from pathlib import Path from urllib.parse import urlparse import httpx from PIL import Image from app.config import settings logger = logging.getLogger(__name__) MAX_IMAGE_FETCH_BYTES = 20 * 1024 * 1024 # 20 MB cap for remote image downloads async def fetch_image_bytes(url: str, timeout: float = 15.0) -> bytes: """Download an image from an HTTP(S) URL and return raw bytes. Performs server-side (no browser CORS), with scheme/size/content-type guards suitable for untrusted user-supplied URLs in a POC context. Raises ``ValueError`` on any failure so the caller can surface a 400. """ parsed = urlparse(url) if parsed.scheme not in ("http", "https"): raise ValueError(f"지원하지 않는 URL 스킴입니다: {parsed.scheme or '(empty)'}") if not parsed.netloc: raise ValueError("URL 호스트를 해석할 수 없습니다.") try: async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client: resp = await client.get(url) except httpx.RequestError as e: raise ValueError(f"이미지를 가져오지 못했습니다: {e}") from e if resp.status_code >= 400: raise ValueError(f"이미지를 가져오지 못했습니다: HTTP {resp.status_code}") content = resp.content if len(content) == 0: raise ValueError("빈 이미지 응답입니다.") if len(content) > MAX_IMAGE_FETCH_BYTES: raise ValueError( f"이미지 크기가 제한을 초과했습니다: {len(content)} > " f"{MAX_IMAGE_FETCH_BYTES} bytes" ) ctype = resp.headers.get("content-type", "").split(";")[0].strip().lower() # Some CDNs return application/octet-stream for images — allow that too. if ctype and not (ctype.startswith("image/") or ctype == "application/octet-stream"): raise ValueError(f"이미지가 아닌 콘텐츠입니다: content-type={ctype}") return content def save_original_image(image_bytes: bytes, user_id: uuid.UUID) -> str: """Save original uploaded image and return path relative to UPLOAD_DIR.""" rel_path = Path("originals") / str(user_id) dest_dir = Path(settings.UPLOAD_DIR) / rel_path dest_dir.mkdir(parents=True, exist_ok=True) filename = f"{uuid.uuid4().hex}.jpg" (dest_dir / filename).write_bytes(image_bytes) return str(rel_path / filename) def save_ghost_mannequin_image(image_bytes: bytes, user_id: uuid.UUID) -> str: """Save ghost mannequin image and return path relative to UPLOAD_DIR.""" rel_path = Path("ghost_mannequin") / str(user_id) dest_dir = Path(settings.UPLOAD_DIR) / rel_path dest_dir.mkdir(parents=True, exist_ok=True) filename = f"{uuid.uuid4().hex}.png" (dest_dir / filename).write_bytes(image_bytes) return str(rel_path / filename) def save_vton_image(image_bytes: bytes, user_id: uuid.UUID) -> str: """Save VTON composite result image and return path relative to UPLOAD_DIR.""" rel_path = Path("vton") / str(user_id) dest_dir = Path(settings.UPLOAD_DIR) / rel_path dest_dir.mkdir(parents=True, exist_ok=True) filename = f"{uuid.uuid4().hex}.png" (dest_dir / filename).write_bytes(image_bytes) return str(rel_path / filename) def save_segmentation_image(image_bytes: bytes) -> str: """Save segmentation result image and return path relative to UPLOAD_DIR.""" rel_path = Path("segmentation") dest_dir = Path(settings.UPLOAD_DIR) / rel_path dest_dir.mkdir(parents=True, exist_ok=True) filename = f"{uuid.uuid4().hex}.png" (dest_dir / filename).write_bytes(image_bytes) return str(rel_path / filename) def delete_image(relative_path: str | None) -> None: """Delete an image file by its path relative to UPLOAD_DIR. Logs a warning and continues if the file doesn't exist or deletion fails. """ if not relative_path: return filepath = Path(settings.UPLOAD_DIR) / relative_path try: filepath.unlink(missing_ok=True) logger.info("Deleted image: %s", relative_path) except OSError: logger.warning("Failed to delete image: %s", relative_path, exc_info=True) def delete_user_storage(user_id: uuid.UUID) -> int: """Delete all storage files for a user across all subdirectories. Returns the number of files deleted. """ count = 0 base = Path(settings.UPLOAD_DIR) for sub in ("originals", "ghost_mannequin", "faces", "vton"): user_dir = base / sub / str(user_id) if user_dir.is_dir(): for f in user_dir.iterdir(): if f.is_file(): try: f.unlink() count += 1 except OSError: logger.warning("Failed to delete: %s", f, exc_info=True) try: user_dir.rmdir() except OSError: pass return count def resize_image_if_needed(image_bytes: bytes, max_size: int = 1536) -> bytes: """Resize image if larger than max_size on any side. Returns JPEG bytes. RGBA inputs (e.g., pre-segmented person PNGs) are composited onto white before the JPEG encode — a plain ``convert("RGB")`` would drop alpha onto a black canvas, which downstream image models then preserve as a black backdrop in the result. """ img = Image.open(io.BytesIO(image_bytes)) if img.mode in ("RGBA", "LA") or (img.mode == "P" and "transparency" in img.info): rgba = img.convert("RGBA") bg = Image.new("RGB", rgba.size, (255, 255, 255)) bg.paste(rgba, mask=rgba.split()[-1]) img = bg elif img.mode != "RGB": img = img.convert("RGB") w, h = img.size if max(w, h) > max_size: ratio = max_size / max(w, h) img = img.resize((int(w * ratio), int(h * ratio)), Image.LANCZOS) buf = io.BytesIO() img.save(buf, format="JPEG", quality=90) return buf.getvalue() def resize_png_if_needed(image_bytes: bytes, max_size: int = 1024) -> bytes: """Resize PNG if larger than max_size on any side, preserving transparency. Returns the original bytes if no resize needed; otherwise returns re-encoded PNG bytes with the alpha channel intact (required for ghost mannequin images so the model can distinguish garment from background). """ img = Image.open(io.BytesIO(image_bytes)) w, h = img.size if max(w, h) <= max_size: return image_bytes ratio = max_size / max(w, h) new_size = (int(w * ratio), int(h * ratio)) img = img.resize(new_size, Image.LANCZOS) buf = io.BytesIO() img.save(buf, format="PNG", optimize=True) return buf.getvalue()