Frank Schwenk
dec74a46d7
On-demand ?w= resize with a disk cache under webcache/; downloads stay full-resolution. Co-authored-by: Cursor <cursoragent@cursor.com>
90 lines
3.0 KiB
Python
90 lines
3.0 KiB
Python
"""On-demand web image resizing with disk cache (srcset helpers)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import tempfile
|
|
from pathlib import Path
|
|
from urllib.parse import quote
|
|
|
|
from PIL import Image
|
|
|
|
from . import config
|
|
|
|
logger = logging.getLogger("livef12.images")
|
|
|
|
# Allowed ?w= values for /jobs/.../files/... — keep in sync with templates.
|
|
SRCSET_WIDTHS: tuple[int, ...] = (320, 640, 960, 1280)
|
|
|
|
# Default sizes for the 1/2/4-column job + variant grids.
|
|
GRID_SIZES = "(min-width: 960px) 25vw, (min-width: 640px) 50vw, 100vw"
|
|
# Original/rembg pair is always two columns.
|
|
PAIR_SIZES = "(min-width: 640px) 50vw, 50vw"
|
|
|
|
|
|
def file_url(job_id: str, filename: str, width: int | None = None) -> str:
|
|
base = f"/jobs/{quote(job_id, safe='')}/files/{quote(filename, safe='')}"
|
|
if width is None:
|
|
return base
|
|
return f"{base}?w={width}"
|
|
|
|
|
|
def srcset_for(job_id: str, filename: str) -> str:
|
|
return ", ".join(f"{file_url(job_id, filename, w)} {w}w" for w in SRCSET_WIDTHS)
|
|
|
|
|
|
def img_attrs(job_id: str, filename: str, *, sizes: str = GRID_SIZES) -> dict[str, str]:
|
|
"""Attrs for responsive <img>: src, srcset, sizes, full_src (lightbox)."""
|
|
return {
|
|
"src": file_url(job_id, filename, 640),
|
|
"srcset": srcset_for(job_id, filename),
|
|
"sizes": sizes,
|
|
# Lightbox: large webp, not the multi-MB master PNG/JPEG.
|
|
"full_src": file_url(job_id, filename, SRCSET_WIDTHS[-1]),
|
|
}
|
|
|
|
|
|
def get_or_create_resized(source: Path, width: int) -> Path:
|
|
"""Return a cached WebP whose longest edge is at most `width` (no upscale)."""
|
|
if width not in SRCSET_WIDTHS:
|
|
raise ValueError(f"unsupported width: {width}")
|
|
if not source.is_file():
|
|
raise FileNotFoundError(source)
|
|
|
|
config.WEB_CACHE_DIR.mkdir(parents=True, exist_ok=True)
|
|
cache_path = config.WEB_CACHE_DIR / f"{source.name}.w{width}.webp"
|
|
try:
|
|
if cache_path.is_file() and cache_path.stat().st_mtime >= source.stat().st_mtime:
|
|
return cache_path
|
|
except OSError:
|
|
pass
|
|
|
|
_write_resized(source, cache_path, width)
|
|
return cache_path
|
|
|
|
|
|
def _write_resized(source: Path, dest: Path, width: int) -> None:
|
|
with Image.open(source) as img:
|
|
img.load()
|
|
if img.mode == "P":
|
|
img = img.convert("RGBA" if "transparency" in img.info else "RGB")
|
|
elif img.mode not in ("RGB", "RGBA"):
|
|
img = img.convert("RGB")
|
|
|
|
if img.width > width:
|
|
new_h = max(1, round(img.height * (width / img.width)))
|
|
img = img.resize((width, new_h), Image.Resampling.LANCZOS)
|
|
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
with tempfile.NamedTemporaryFile(
|
|
dir=dest.parent, suffix=".webp", delete=False
|
|
) as tmp:
|
|
tmp_path = Path(tmp.name)
|
|
try:
|
|
img.save(tmp_path, format="WEBP", quality=78, method=4)
|
|
tmp_path.replace(dest)
|
|
except Exception:
|
|
tmp_path.unlink(missing_ok=True)
|
|
raise
|
|
logger.debug("cached %s -> %s (%dx%d)", source.name, dest.name, img.width, img.height)
|