| """
|
| Self-hosted image upscaler for crisp HD/4K hero art.
|
|
|
| AniList's CDN serves covers at max 460x690 and banners at max 1900x400,
|
| and free upscale proxies (wsrv.nl / weserv / statically) block
|
| s4.anilist.co. So we do the upscale ourselves with Pillow (LANCZOS +
|
| unsharp mask) and cache the result on disk under /img_cache, then serve it
|
| with long Cache-Control headers. Only allowed image hosts are accepted.
|
| """
|
| import hashlib
|
| import io
|
| import logging
|
| import os
|
| import threading
|
| import time
|
| from urllib.parse import urlparse
|
|
|
| import requests
|
|
|
| from PIL import Image, ImageFilter, ImageOps
|
|
|
| logger = logging.getLogger(__name__)
|
|
|
| ALLOWED_IMAGE_DOMAINS = (
|
| "s4.anilist.co",
|
| "i.anilist.co",
|
| "cdn.myanimelist.net",
|
| "media.kitsu.io",
|
| "image.tmdb.org",
|
| "m.media-amazon.com",
|
| )
|
|
|
|
|
| _CACHE_DIR = os.path.join(
|
| os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
|
| "img_cache",
|
| )
|
| _CACHE_MAX_BYTES = 768 * 1024 * 1024
|
| _CACHE_MAX_AGE = 30 * 24 * 3600
|
| _FETCH_TIMEOUT = 20
|
| _LOCK = threading.Lock()
|
|
|
| _FETCH_HEADERS = {
|
| "User-Agent": (
|
| "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:150.0) "
|
| "Gecko/20100101 Firefox/150.0"
|
| ),
|
| "Accept": "image/avif,image/webp,image/*,*/*;q=0.8",
|
| }
|
|
|
|
|
| def is_allowed_image_url(url: str) -> bool:
|
| """True when the URL points at a trusted image host."""
|
| try:
|
| parsed = urlparse(url)
|
| except ValueError:
|
| return False
|
| if parsed.scheme not in ("http", "https"):
|
| return False
|
| host = (parsed.hostname or "").lower()
|
| return host in ALLOWED_IMAGE_DOMAINS
|
|
|
|
|
| def _cache_key(url: str, width: int) -> str:
|
| digest = hashlib.sha1(f"{url}|{width}".encode("utf-8")).hexdigest()
|
| return digest
|
|
|
|
|
| def _cache_path(key: str) -> str:
|
| return os.path.join(_CACHE_DIR, key + ".jpg")
|
|
|
|
|
| def _read_cache(key: str):
|
| path = _cache_path(key)
|
| try:
|
| if not os.path.exists(path):
|
| return None
|
| if time.time() - os.path.getmtime(path) > _CACHE_MAX_AGE:
|
| return None
|
| with open(path, "rb") as f:
|
| return f.read()
|
| except OSError:
|
| return None
|
|
|
|
|
| def _write_cache(key: str, data: bytes):
|
| try:
|
| os.makedirs(_CACHE_DIR, exist_ok=True)
|
| tmp_path = _cache_path(key) + ".tmp"
|
| with open(tmp_path, "wb") as f:
|
| f.write(data)
|
| os.replace(tmp_path, _cache_path(key))
|
| except OSError as e:
|
| logger.warning("Image cache write failed: %s", e)
|
|
|
|
|
| def _prune_cache():
|
| """Delete oldest files until the cache fits under _CACHE_MAX_BYTES."""
|
| try:
|
| entries = []
|
| for name in os.listdir(_CACHE_DIR):
|
| path = os.path.join(_CACHE_DIR, name)
|
| try:
|
| entries.append((os.path.getmtime(path), os.path.getsize(path), path))
|
| except OSError:
|
| continue
|
| total = sum(sz for _, sz, _ in entries)
|
| if total <= _CACHE_MAX_BYTES:
|
| return
|
| entries.sort()
|
| for _mtime, size, path in entries:
|
| if total <= _CACHE_MAX_BYTES:
|
| break
|
| try:
|
| os.remove(path)
|
| total -= size
|
| except OSError:
|
| pass
|
| except OSError:
|
| pass
|
|
|
|
|
| def _fetch_image(url: str) -> bytes:
|
| resp = requests.get(url, headers=_FETCH_HEADERS, timeout=_FETCH_TIMEOUT)
|
| resp.raise_for_status()
|
| return resp.content
|
|
|
|
|
| def upscale_image(url: str, width: int = 1920, quality: int = 88) -> bytes:
|
| """
|
| Fetch url, upscale it to `width` px (LANCZOS + unsharp mask), cache on
|
| disk and return the JPEG bytes. If the source is already wider than the
|
| requested width, it is returned as-is (no pointless upscale).
|
| """
|
| width = min(max(int(width), 480), 3840)
|
| key = _cache_key(url, width)
|
|
|
| cached = _read_cache(key)
|
| if cached is not None:
|
| return cached
|
|
|
| with _LOCK:
|
| cached = _read_cache(key)
|
| if cached is not None:
|
| return cached
|
|
|
| raw = _fetch_image(url)
|
|
|
| try:
|
| img = Image.open(io.BytesIO(raw))
|
| img = ImageOps.exif_transpose(img)
|
| if img.mode not in ("RGB", "L"):
|
| img = img.convert("RGB")
|
|
|
| if img.width < width:
|
| target_h = max(1, round(img.height * width / img.width))
|
| img = img.resize((width, target_h), Image.LANCZOS)
|
|
|
| img = img.filter(ImageFilter.UnsharpMask(radius=2, percent=110, threshold=2))
|
|
|
| out = io.BytesIO()
|
| img.save(out, format="JPEG", quality=quality, optimize=True, progressive=True)
|
| result = out.getvalue()
|
| except Exception as e:
|
| logger.warning("Image upscale failed (%s), serving original: %s", url, e)
|
| result = raw
|
|
|
| _write_cache(key, result)
|
| _prune_cache()
|
| return result
|
|
|
