""" 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 location: /img_cache _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 # 768 MB cap _CACHE_MAX_AGE = 30 * 24 * 3600 # 30 days _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) # Light unsharp mask makes the upscale look noticeably crisper 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