mtga-assistant/src/mtga_assistant/images.py

"""Card images from Scryfall with a size-capped on-disk LRU cache.

Images are fetched lazily, only for cards that actually show up, and stored as
``<cache>/images/<grpId>.jpg``. A file's mtime is bumped whenever it is used, and
the least recently used files are evicted when the cache exceeds its cap.
"""

from __future__ import annotations

import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

from PySide6.QtCore import QObject, Signal
from PySide6.QtGui import QPixmap

from .carddb import CardDB

USER_AGENT = "mtga-assistant/0.1 (personal desktop tracker)"
SCRYFALL = "https://api.scryfall.com"
MIN_REQUEST_INTERVAL = 0.11  # Scryfall asks for <= 10 requests/second


class ImageCache(QObject):
    image_ready = Signal(int)  # grpId
    size_changed = Signal(int)  # bytes on disk
    _downloaded = Signal(int)

    def __init__(self, cache_dir: Path, cards: CardDB, max_bytes: int):
        super().__init__()
        self.dir = cache_dir / "images"
        self.dir.mkdir(parents=True, exist_ok=True)
        self.cards = cards
        self.max_bytes = max_bytes
        self._pool = ThreadPoolExecutor(max_workers=3, thread_name_prefix="scryfall")
        self._lock = threading.Lock()
        self._inflight: set[int] = set()
        self._failed: set[int] = set()
        self._last_request = 0.0
        self._pixmaps: dict[int, QPixmap] = {}
        # Emitted from worker threads; Qt queues delivery onto this object's (GUI) thread.
        self._downloaded.connect(self._on_downloaded)
        self.evict()

    def path_for(self, grp_id: int) -> Path:
        # Tokens get their own names: early versions fetched wrong images for them.
        prefix = "t" if self.cards.get(grp_id).is_token else ""
        return self.dir / f"{prefix}{grp_id}.jpg"

    def pixmap(self, grp_id: int) -> QPixmap | None:
        """Return the image if cached; otherwise start a download and return None."""
        if pm := self._pixmaps.get(grp_id):
            return pm
        path = self.path_for(grp_id)
        if path.exists():
            try:
                path.touch()
            except OSError:
                pass
            pm = QPixmap(str(path))
            if not pm.isNull():
                if len(self._pixmaps) > 200:
                    self._pixmaps.clear()
                self._pixmaps[grp_id] = pm
                return pm
        self.request(grp_id)
        return None

    def request(self, grp_id: int) -> None:
        with self._lock:
            if grp_id in self._inflight or grp_id in self._failed:
                return
            self._inflight.add(grp_id)
        self._pool.submit(self._download, grp_id)

    def is_failed(self, grp_id: int) -> bool:
        return grp_id in self._failed

    # -- worker side ---------------------------------------------------------

    def _throttle(self) -> None:
        with self._lock:
            wait = self._last_request + MIN_REQUEST_INTERVAL - time.monotonic()
            self._last_request = max(time.monotonic(), self._last_request + MIN_REQUEST_INTERVAL)
        if wait > 0:
            time.sleep(wait)

    def _candidate_urls(self, grp_id: int) -> list[str]:
        q = "format=image&version=normal"
        info = self.cards.get(grp_id)
        if info.is_token:
            # MTGA token ids/numbers don't line up with Scryfall's cards; tokens live in "t<set>".
            name = urllib.parse.quote(info.name)
            urls = []
            if info.set_code:
                urls.append(f"{SCRYFALL}/cards/named?exact={name}&set=t{urllib.parse.quote(info.set_code.lower())}&{q}")
            urls.append(f"{SCRYFALL}/cards/search?q={urllib.parse.quote(f'!"{info.name}" is:token')}&format=image&version=normal")
            return urls
        urls = [f"{SCRYFALL}/cards/arena/{grp_id}?{q}"]
        if info.set_code and info.collector_number:
            set_code = urllib.parse.quote(info.set_code.lower())
            cn = urllib.parse.quote(info.collector_number)
            urls.append(f"{SCRYFALL}/cards/{set_code}/{cn}?{q}")
        if info.name and not info.name.startswith(("Unknown card", "Card #")):
            exact = urllib.parse.quote(info.name.split(" // ")[0])
            urls.append(f"{SCRYFALL}/cards/named?exact={exact}&{q}")
        return urls

    def _download(self, grp_id: int) -> None:
        ok = False
        try:
            for url in self._candidate_urls(grp_id):
                self._throttle()
                req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT, "Accept": "*/*"})
                try:
                    with urllib.request.urlopen(req, timeout=20) as resp:
                        data = resp.read()
                except urllib.error.HTTPError as e:
                    if e.code == 404:
                        continue
                    break
                except (urllib.error.URLError, TimeoutError, OSError):
                    break
                tmp = self.path_for(grp_id).with_suffix(".part")
                tmp.write_bytes(data)
                tmp.replace(self.path_for(grp_id))
                ok = True
                break
        finally:
            with self._lock:
                self._inflight.discard(grp_id)
                if not ok:
                    self._failed.add(grp_id)
        if ok:
            self._downloaded.emit(grp_id)

    def _on_downloaded(self, grp_id: int) -> None:
        self.evict()
        self.image_ready.emit(grp_id)

    # -- cache management ------------------------------------------------------

    def disk_usage(self) -> int:
        return sum(p.stat().st_size for p in self.dir.glob("*.jpg"))

    def evict(self) -> None:
        files = []
        for p in self.dir.glob("*.jpg"):
            try:
                st = p.stat()
            except OSError:
                continue
            files.append((st.st_mtime, st.st_size, p))
        total = sum(f[1] for f in files)
        if total > self.max_bytes:
            for _, size, p in sorted(files):
                try:
                    p.unlink()
                except OSError:
                    continue
                total -= size
                if p.stem.lstrip("t").isdigit():
                    self._pixmaps.pop(int(p.stem.lstrip("t")), None)
                if total <= self.max_bytes * 0.9:
                    break
        self.size_changed.emit(total)

    def clear(self) -> None:
        for p in self.dir.glob("*"):
            try:
                p.unlink()
            except OSError:
                pass
        self._pixmaps.clear()
        self._failed.clear()
        self.size_changed.emit(0)

    def shutdown(self) -> None:
        self._pool.shutdown(wait=False, cancel_futures=True)