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
import time
from typing import TYPE_CHECKING, Final

import httpx

from signal_bot.analytics import probe

if TYPE_CHECKING:
    from collections.abc import Sequence


MS_PER_SECOND: Final = 1000


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 base64-encoded, ready for Signal."""
        return base64.b64encode(await self.fetch_bytes(api_url, url_path, rewrite)).decode("ascii")

    async def fetch_bytes(
        self,
        api_url: str,
        url_path: Sequence[str | int],
        rewrite: tuple[str, str] | None = None,
        params: dict[str, str] | None = None,
    ) -> bytes:
        """Fetch an image and return its raw bytes.

        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.

            params: Query string for the JSON endpoint, when it is easier to
                pass than to embed -- the Commons API takes a title that has
                to be percent-encoded, the animal APIs are fixed URLs.

        Returns:
            The image bytes.

        Raises:
            ImageFetchError: The JSON did not contain a usable URL.
        """
        image_url = _walk(await self.fetch_json(api_url, params), url_path)
        if rewrite is not None:
            image_url = image_url.replace(*rewrite)
        return await self.fetch_image(image_url)

    async def _get(self, url: str, params: dict[str, str] | None = None) -> httpx.Response:
        """Make one request, time it, and report it to whatever is probing.

        The single place this class touches the network, which is what makes
        "how often does thecatapi fail, and how slow is Commons" answerable
        without a change to any command. ``probe.note_upstream`` is a no-op
        when nothing is measuring, so the offline runs and the tests are
        unaffected.

        Only the hostname is reported. A URL carries identifiers -- the poem
        one carries a Wolne Lektury slug -- and identifiers belong in
        ``item_id``, where a deliberate decision governs them.

        A request that never got an answer is recorded as NO_RESPONSE and
        re-raised unchanged: DNS, connect and timeout failures are the ones
        that never reach a status code, and losing them would make an
        upstream that is entirely down look merely unused.
        """
        started = time.perf_counter()
        host = httpx.URL(url).host
        try:
            response = await self._client.get(url, params=params)
        except httpx.HTTPError:
            probe.note_upstream(host, probe.NO_RESPONSE, _elapsed_ms(started))
            raise
        probe.note_upstream(host, response.status_code, _elapsed_ms(started))
        response.raise_for_status()
        return response

    async def fetch_json(self, api_url: str, params: dict[str, str] | None = None) -> object:
        """Fetch one JSON document from an API.

        Separate from fetch_bytes because a Commons answer carries more than
        the picture: !pułapka needs the description in the same reply, and
        asking twice would be both wasteful and a chance for the two halves
        to disagree.
        """
        return (await self._get(api_url, params)).json()

    async def fetch_text(self, url: str) -> str:
        """Fetch one plain-text document.

        !wiersz is the only command whose upstream serves neither JSON nor an
        image: Wolne Lektury hands back the poem as text/plain. It goes
        through this client like everything else, so the offline environment
        and the tests serve it the same way they serve a cat.
        """
        return (await self._get(url)).text

    async def fetch_image(self, image_url: str) -> bytes:
        """Fetch the bytes at an already-resolved image URL.

        These bytes are never written to a path built from the remote URL:
        the original bot named a local file after the URL's last path
        segment and wrote it into the process CWD, which is both path
        traversal and unbounded. portraits.py does cache to disk, but under
        a name it derives from the pope's ordinal, never from upstream.
        """
        return (await self._get(image_url)).content


def _elapsed_ms(started: float) -> int:
    """Milliseconds since a perf_counter reading."""
    return round((time.perf_counter() - started) * MS_PER_SECOND)


_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