36 lines
1.1 KiB
Python
36 lines
1.1 KiB
Python
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
|