api/utils/image_upscale.py

"""
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