signal-bot/signal_bot/animals.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""The image commands, and the alias tables people actually type."""

from __future__ import annotations

import json
import secrets
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Final

from signal_bot.config import DEFAULT_CAT_API_URL, DEFAULT_DOG_API_URL, DEFAULT_GECKO_API_URL

if TYPE_CHECKING:
    from collections.abc import Sequence

    from signal_bot.decorators import Decorator
    from signal_bot.http_client import ImageFetcher
    from signal_bot.registry import CommandContext, CommandRegistry, Handler

ALIAS_FILE: Final = Path(__file__).with_name("aliases.json")


@dataclass(frozen=True, slots=True)
class ImageSource:
    """Where one image command gets its picture, and how to read the answer."""

    api_url: str
    url_path: tuple[str | int, ...]
    rewrite: tuple[str, str] | None = None
    page_limit: int | None = None


# iNaturalist's pagination ceiling: page * per_page must stay at or under
# 10 000, and page=10001 comes back empty. Measured, not documented.
INATURALIST_PAGE_LIMIT: Final = 10_000

# Two iNaturalist quirks, kept beside the endpoint because that is what they
# are properties of. It answers with a 75px square thumbnail whose full-size
# sibling is one path segment away; and it accepts `order_by=random` while
# ignoring it -- five requests in a row returned the same gecko -- so the
# randomness has to come from asking for a random page instead.
SOURCES: Final[dict[str, ImageSource]] = {
    "cat": ImageSource(DEFAULT_CAT_API_URL, (0, "url")),
    "dog": ImageSource(DEFAULT_DOG_API_URL, ("message",)),
    "gecko": ImageSource(
        DEFAULT_GECKO_API_URL,
        ("results", 0, "photos", 0, "url"),
        ("/square.", "/large."),
        INATURALIST_PAGE_LIMIT,
    ),
}


def paged_url(api_url: str, page_limit: int | None) -> str:
    """Return api_url, pointed at a random page when the source needs one.

    secrets rather than random: the choice is not security-sensitive, but
    ruff's ALL rule set flags random for exactly this shape, and this repo
    forbids the noqa that would silence it.
    """
    if page_limit is None:
        return api_url
    separator = "&" if "?" in api_url else "?"
    return f"{api_url}{separator}page={secrets.randbelow(page_limit) + 1}"


def load_aliases(path: Path = ALIAS_FILE) -> dict[str, list[str]]:
    """Load the trigger table.

    Aliases live in JSON rather than in source so that the styled Unicode
    variants are data the linter never has to reason about.
    """
    document = json.loads(path.read_text(encoding="utf-8"))
    return {name: list(triggers) for name, triggers in document.items() if not name.startswith("_")}


def image_handler(
    fetcher: ImageFetcher,
    api_url: str,
    url_path: Sequence[str | int],
    rewrite: tuple[str, str] | None = None,
    page_limit: int | None = None,
) -> Handler:
    """Build a handler that posts one random image from a JSON image API."""

    async def handler(ctx: CommandContext) -> None:
        url = paged_url(api_url, page_limit)
        encoded = await fetcher.fetch_base64(url, url_path, rewrite)
        await ctx.reply_image(encoded)

    return handler


def register_animal_commands(
    registry: CommandRegistry,
    fetcher: ImageFetcher,
    *,
    decorate: Decorator | None = None,
    aliases: dict[str, list[str]] | None = None,
    endpoints: dict[str, str] | None = None,
) -> None:
    """Register every image command on a registry.

    Args:
        registry: Registry to populate.
        fetcher: Injected image fetcher.
        decorate: Optional callable applied to each handler, used to attach
            the cooldown gate without hard-wiring policy into the commands.
        aliases: Trigger table. Defaults to the shipped aliases.json.
        endpoints: Command name to image-API URL. Defaults to the real
            public APIs; the offline environment points these at itself.
    """
    table = load_aliases() if aliases is None else aliases
    overrides = endpoints or {}
    for name, source in SOURCES.items():
        api_url = overrides.get(name, source.api_url)
        handler = image_handler(
            fetcher, api_url, source.url_path, source.rewrite, source.page_limit
        )
        if decorate is not None:
            handler = decorate(handler)
        registry.register(name, table[name], handler)