feat: image download, resize-to-fit, baseline JPEG conversion
Made-with: Cursor
This commit is contained in:
61
src/images.py
Normal file
61
src/images.py
Normal file
@@ -0,0 +1,61 @@
|
||||
import hashlib
|
||||
import logging
|
||||
import os
|
||||
from io import BytesIO
|
||||
|
||||
import requests
|
||||
from PIL import Image as PILImage
|
||||
|
||||
import config
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _url_hash(url: str) -> str:
|
||||
return hashlib.sha256(url.encode()).hexdigest()[:16]
|
||||
|
||||
|
||||
def _resize_to_fit(img: PILImage.Image) -> PILImage.Image:
|
||||
w, h = img.size
|
||||
if w >= h:
|
||||
max_w, max_h = config.IMAGE_MAX_LANDSCAPE
|
||||
else:
|
||||
max_w, max_h = config.IMAGE_MAX_PORTRAIT
|
||||
|
||||
scale = min(max_w / w, max_h / h)
|
||||
|
||||
new_w = int(w * scale)
|
||||
new_h = int(h * scale)
|
||||
|
||||
if new_w == w and new_h == h:
|
||||
return img
|
||||
|
||||
return img.resize((new_w, new_h), PILImage.Resampling.LANCZOS)
|
||||
|
||||
|
||||
def process_image(url: str, output_dir: str) -> tuple[str, int, int]:
|
||||
"""Download an image, resize it, save as baseline JPEG.
|
||||
|
||||
Returns (local_path, width, height). Deduplicates by URL hash.
|
||||
"""
|
||||
os.makedirs(output_dir, exist_ok=True)
|
||||
filename = f"{_url_hash(url)}.jpg"
|
||||
local_path = os.path.join(output_dir, filename)
|
||||
|
||||
if os.path.exists(local_path):
|
||||
img = PILImage.open(local_path)
|
||||
return local_path, img.width, img.height
|
||||
|
||||
response = requests.get(url, timeout=30)
|
||||
response.raise_for_status()
|
||||
|
||||
img = PILImage.open(BytesIO(response.content))
|
||||
|
||||
if img.mode in ("RGBA", "P", "LA"):
|
||||
img = img.convert("RGB")
|
||||
|
||||
img = _resize_to_fit(img)
|
||||
|
||||
img.save(local_path, format="JPEG", progressive=False, quality=85)
|
||||
|
||||
return local_path, img.width, img.height
|
||||
80
tests/test_images.py
Normal file
80
tests/test_images.py
Normal file
@@ -0,0 +1,80 @@
|
||||
import os
|
||||
import pytest
|
||||
from io import BytesIO
|
||||
from unittest.mock import patch, MagicMock
|
||||
from PIL import Image as PILImage
|
||||
from src.images import process_image, _resize_to_fit
|
||||
|
||||
|
||||
def _make_test_image(width, height, fmt="JPEG"):
|
||||
img = PILImage.new("RGB", (width, height), color="red")
|
||||
buf = BytesIO()
|
||||
img.save(buf, format=fmt)
|
||||
buf.seek(0)
|
||||
return buf.read()
|
||||
|
||||
|
||||
def test_resize_landscape_downscale():
|
||||
img = PILImage.new("RGB", (1600, 900))
|
||||
result = _resize_to_fit(img)
|
||||
assert result.width <= 800
|
||||
assert result.height <= 480
|
||||
assert result.width / result.height == pytest.approx(1600 / 900, rel=0.02)
|
||||
|
||||
|
||||
def test_resize_portrait_downscale():
|
||||
img = PILImage.new("RGB", (600, 1200))
|
||||
result = _resize_to_fit(img)
|
||||
assert result.width <= 480
|
||||
assert result.height <= 800
|
||||
assert result.width / result.height == pytest.approx(600 / 1200, rel=0.02)
|
||||
|
||||
|
||||
def test_resize_small_image_upscales():
|
||||
img = PILImage.new("RGB", (200, 100))
|
||||
result = _resize_to_fit(img)
|
||||
assert result.width > 200
|
||||
assert result.width <= 800
|
||||
assert result.height <= 480
|
||||
|
||||
|
||||
def test_resize_already_fits():
|
||||
img = PILImage.new("RGB", (800, 480))
|
||||
result = _resize_to_fit(img)
|
||||
assert result.size == (800, 480)
|
||||
|
||||
|
||||
def test_process_image_downloads_and_saves(tmp_path):
|
||||
image_bytes = _make_test_image(1024, 768)
|
||||
|
||||
mock_response = MagicMock()
|
||||
mock_response.content = image_bytes
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
|
||||
with patch("src.images.requests.get", return_value=mock_response):
|
||||
path, w, h = process_image(
|
||||
"https://example.com/photo.jpg", str(tmp_path)
|
||||
)
|
||||
|
||||
assert os.path.exists(path)
|
||||
assert path.endswith(".jpg")
|
||||
assert w <= 800
|
||||
assert h <= 480
|
||||
|
||||
saved = PILImage.open(path)
|
||||
assert saved.format == "JPEG"
|
||||
|
||||
|
||||
def test_process_image_dedup(tmp_path):
|
||||
image_bytes = _make_test_image(500, 300)
|
||||
|
||||
mock_response = MagicMock()
|
||||
mock_response.content = image_bytes
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
|
||||
with patch("src.images.requests.get", return_value=mock_response) as mock_get:
|
||||
path1, _, _ = process_image("https://example.com/same.jpg", str(tmp_path))
|
||||
path2, _, _ = process_image("https://example.com/same.jpg", str(tmp_path))
|
||||
|
||||
assert path1 == path2
|
||||
assert mock_get.call_count == 1
|
||||
Reference in New Issue
Block a user