signal-bot/signal_bot/portraits.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""On-disk cache of papal portraits, keyed by ordinal.

The set is closed: 267 popes, one portrait each, so this cache is bounded by
construction and never evicts. That is what makes writing images to disk safe
here when http_client refuses to -- the filename comes from the ordinal, so
nothing upstream controls a path, and the directory belongs to systemd.
"""

from __future__ import annotations

import base64
import os
from pathlib import Path
from typing import TYPE_CHECKING

from signal_bot.commons import SUFFIXES, THUMB_PATH, imageinfo_params
from signal_bot.endpoints import THUMB_WIDTH
from signal_bot.http_client import ImageFetchError

if TYPE_CHECKING:
    from signal_bot.http_client import ImageFetcher
    from signal_bot.versioning import Pope


def default_cache_dir() -> Path:
    """Return the portrait directory, preferring the one systemd hands us.

    Under the unit CACHE_DIRECTORY is set and writable even though
    ProtectHome=read-only makes the rest of $HOME immutable; outside it -- a
    test run, a shell -- XDG applies. Mirrors scripts/signal_api_watchdog.sh.
    """
    explicit = os.environ.get("PORTRAIT_CACHE_DIR", "").strip()
    if explicit:
        return Path(explicit)
    cache = os.environ.get("CACHE_DIRECTORY", "").strip()
    if cache:
        return Path(cache) / "portraits"
    home = os.environ.get("XDG_CACHE_HOME", "").strip() or str(Path.home() / ".cache")
    return Path(home) / "signal-bot" / "portraits"


class PortraitCache:
    """Fetches a pope's portrait once, then serves it from disk forever."""

    def __init__(
        self,
        fetcher: ImageFetcher,
        api_url: str,
        cache_dir: Path | None = None,
    ) -> None:
        """Store the fetcher, the Commons endpoint and where files land."""
        self._fetcher = fetcher
        self._api_url = api_url
        self._dir = default_cache_dir() if cache_dir is None else cache_dir

    def path_for(self, pope: Pope) -> Path:
        """Return the cache path for one pope.

        Named from the ordinal, never from the Commons filename: those carry
        spaces, parentheses, commas and non-ASCII, and are attacker-adjacent
        data in a way an integer is not.
        """
        suffix = Path(pope.portrait).suffix.lower()
        if suffix not in SUFFIXES:
            msg = f"portrait for #{pope.ordinal} has an unusable suffix: {pope.portrait!r}"
            raise ImageFetchError(msg)
        return self._dir / f"{pope.ordinal:03d}{suffix}"

    async def get_base64(self, pope: Pope) -> str:
        """Return the portrait base64-encoded, as the transport wants it."""
        return base64.b64encode(await self.get_bytes(pope)).decode("ascii")

    async def get_bytes(self, pope: Pope) -> bytes:
        """Return the portrait bytes, downloading them only on a miss."""
        path = self.path_for(pope)
        cached = _read(path)
        if cached is not None:
            return cached

        image = await self._fetcher.fetch_bytes(
            self._api_url,
            THUMB_PATH,
            params=imageinfo_params(THUMB_WIDTH, title=f"File:{pope.portrait}"),
        )
        _write(path, image)
        return image


def _read(path: Path) -> bytes | None:
    """Return cached bytes, or None when there is nothing usable there.

    An empty file is a miss, not a hit: that is what a crash between create
    and write leaves behind, and serving it would post a broken attachment
    into the group forever.
    """
    try:
        data = path.read_bytes()
    except OSError:
        return None
    return data or None


def _write(path: Path, image: bytes) -> None:
    """Write the portrait atomically, ignoring a cache that will not take it.

    A read-only or missing cache directory must not cost the group its photo
    -- the bytes are already in hand. The next call simply misses again.
    """
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        temporary = path.with_suffix(path.suffix + ".tmp")
        temporary.write_bytes(image)
        temporary.replace(path)
    except OSError:
        return