signal-bot/signal_bot/http_client.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Outbound HTTP for the image commands.

Every outbound request goes through this one injectable client, so the tests
and the offline environment can serve fixtures without the bot knowing.
"""

from __future__ import annotations

import base64
from typing import TYPE_CHECKING, Final

if TYPE_CHECKING:
    from collections.abc import Sequence

    import httpx


class ImageFetchError(RuntimeError):
    """Raised when an upstream image API did not yield a usable image."""


class ImageFetcher:
    """Resolves an image URL from a JSON API and returns it base64-encoded."""

    def __init__(self, client: httpx.AsyncClient) -> None:
        """Store the injected HTTP client."""
        self._client = client

    async def fetch_base64(
        self,
        api_url: str,
        url_path: Sequence[str | int],
        rewrite: tuple[str, str] | None = None,
    ) -> str:
        """Fetch an image and return it as base64.

        Args:
            api_url: JSON endpoint that names an image.
            url_path: Keys/indices to walk in the JSON to reach the image URL.
            rewrite: Optional (find, replace) pair applied to the URL found
                there. Some APIs name a thumbnail and leave the full-size
                photo one substitution away -- iNaturalist's observation JSON
                gives a 75px `/square.` URL whose `/large.` sibling is the
                real photo -- and a substitution keeps that upstream quirk as
                data next to the endpoint rather than as a branch in here.

        Returns:
            The image bytes, base64-encoded.

        Raises:
            ImageFetchError: The JSON did not contain a usable URL.
        """
        response = await self._client.get(api_url)
        response.raise_for_status()
        image_url = _walk(response.json(), url_path)
        if rewrite is not None:
            image_url = image_url.replace(*rewrite)

        # Never written to disk: the original bot named a local file after the
        # remote URL's last path segment and wrote it into the process CWD.
        image = await self._client.get(image_url)
        image.raise_for_status()
        return base64.b64encode(image.content).decode("ascii")


_MISSING: Final = object()


def _in_range(items: list[object], index: int) -> bool:
    """True when index addresses a real element, counting from either end."""
    return -len(items) <= index < len(items)


def _step(node: object, key: str | int, path: Sequence[str | int]) -> object:
    """Descend one level into a JSON document.

    Narrows explicitly instead of indexing blindly, so a malformed upstream
    response raises ImageFetchError rather than a TypeError from deeper in.
    """
    found: object = _MISSING
    if isinstance(key, str) and isinstance(node, dict):
        found = node.get(key, _MISSING)
    elif isinstance(key, int) and isinstance(node, list) and _in_range(node, key):
        found = node[key]
    if found is _MISSING:
        msg = f"image API response has no element at {path!r}"
        raise ImageFetchError(msg)
    return found


def _walk(document: object, path: Sequence[str | int]) -> str:
    """Walk a JSON document along path and return the string found there."""
    node: object = document
    for key in path:
        node = _step(node, key, path)
    if not isinstance(node, str) or not node:
        msg = f"image API element at {path!r} is not a URL: {node!r}"
        raise ImageFetchError(msg)
    return node