from __future__ import annotations import io from typing import Optional, Tuple import requests from PIL import Image DEFAULT_MAX_BYTES = 50 * 1024 * 1024 # 50MB class ImageLoadError(Exception): pass def fetch_url_bytes(url: str, timeout: float = 10.0, max_bytes: int = DEFAULT_MAX_BYTES) -> bytes: try: with requests.get(url, stream=True, timeout=timeout) as r: r.raise_for_status() buf = io.BytesIO() total = 0 for chunk in r.iter_content(chunk_size=1024 * 64): if not chunk: continue total += len(chunk) if total > max_bytes: raise ImageLoadError(f"Image exceeds max_bytes={max_bytes}") buf.write(chunk) return buf.getvalue() except Exception as e: raise ImageLoadError(f"Cannot fetch image url: {e}") from e def bytes_to_pil(data: bytes) -> Image.Image: try: img = Image.open(io.BytesIO(data)).convert("RGB") return img except Exception as e: raise ImageLoadError(f"Cannot decode image: {e}") from e