File size: 5,016 Bytes
0c0bd98 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 | """
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: <project root>/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
|