signal-bot/signal_bot/pack_command.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""The photo commands a private pack defines, and how one message is answered.

One command per entry, so the cooldown, the log line and the rate-limit row
are per person -- and every one of them is named by the entry's pseudonym
(``pack:p1``), never by what was typed. Reading the journal or the database
tells you somebody asked for a photo, and nothing about whose.

Several names in one message mean "a photo with all of them": the trigger
takes the rest of the line as further names, each resolved through the same
table the trigger came from, and the sets are intersected. Where an entry
offers a choice, the choice is made among the options that actually have a
photo -- picking blindly and then reporting failure would be a "no" the
album could have answered.
"""

from __future__ import annotations

import base64
import secrets
from typing import TYPE_CHECKING, Final

from signal_bot.analytics import probe
from signal_bot.pack import candidate_keysets, media_type

if TYPE_CHECKING:
    from collections.abc import Sequence

    from signal_bot.context import CommandContext
    from signal_bot.decorators import Decorator
    from signal_bot.pack import Entry
    from signal_bot.pack_holder import PackHolder
    from signal_bot.registry import CommandRegistry, Handler

# The prefix every pack command's name carries, so one glance at a log line
# or a command_uses row says where it came from without saying who.
NAME_PREFIX: Final = "pack:"

NO_PHOTO: Final = "BEEP BOOP NIE MAM TAKIEGO ZDJĘCIA."

# What an argument resolver yields for a word it cannot turn into a key. It
# can never collide with a real one: pack_load requires person keys to be
# ascii alphanumeric, and a custom key is "t" followed by digits.
UNKNOWN: Final = "?"

# Deliberately says nothing about who the entry is: !pomoc prints this next
# to the trigger, and the trigger is the only place a name may appear.
ONE_PERSON: Final = "losowe zdjęcie tej osoby"
EVERYBODY: Final = "losowe zdjęcie, na którym są oni wszyscy"
EITHER: Final = "losowe zdjęcie jednego z nich"


def description_for(entry: Entry) -> str:
    """What !pomoc says an entry does, without naming anybody."""
    if len(entry.options) > 1:
        return EITHER
    return ONE_PERSON if len(entry.options[0]) == 1 else EVERYBODY


def pick(items: Sequence[str]) -> str:
    """One item, uniformly at random.

    secrets rather than random for the reason animals.paged_url gives: the
    choice is not security-sensitive, but ruff's ALL rule set flags `random`
    and this repo forbids the noqa that would silence it.
    """
    return items[secrets.randbelow(len(items))]


def pack_handler(holder: PackHolder, entry: Entry) -> Handler:
    """Build the handler that answers one entry's trigger.

    Reads ``holder.pack`` at call time rather than closing over the album it
    was built with: a photo !add_photo stored a second ago has to be visible
    to this person's own command, and a restart is not an acceptable price.
    """

    async def handler(ctx: CommandContext) -> None:
        pack = holder.pack
        wanted = candidate_keysets(pack, entry, ctx.args)
        viable = [keys for keys in wanted or () if pack.matching(keys)]
        if not viable:
            # One line and no photo, for a name nobody has, a combination
            # nobody photographed, and a tail of six names alike: from the
            # group's side those are the same question with the same answer.
            await ctx.reply(NO_PHOTO)
            return
        filename = pick(pack.matching(pick_keys(viable)))
        # The stored name is content-addressed -- pack_store.content_name is
        # the first 128 bits of the file's SHA-256 -- so it identifies the
        # photograph without describing it or anyone in it. That is what
        # makes "which photo keeps coming back" answerable at all.
        probe.note_item(filename)
        payload = base64.b64encode(pack.media_path(filename).read_bytes()).decode("ascii")
        await ctx.reply_media(payload, media_type(filename), filename)

    return handler


def pick_keys(viable: Sequence[frozenset[str]]) -> frozenset[str]:
    """One of the readings that has photos, chosen at random."""
    return viable[secrets.randbelow(len(viable))]


def register_pack_commands(
    registry: CommandRegistry,
    holder: PackHolder,
    *,
    decorate: Decorator | None = None,
) -> None:
    """Register one command per pack entry, scoped to the pack's groups.

    The entries and the groups are read once, here, and never again: a swap
    only ever replaces Pack *data*. If it could add a command it could also
    add one to the wrong group, which is the failure this album exists to
    make impossible.

    Args:
        registry: Registry to populate.
        holder: The album, in the mutable box every handler reads through.
        decorate: Optional callable applied to each handler, used to attach
            the cooldown gate without hard-wiring policy into the commands.
    """
    pack = holder.pack
    for entry in pack.entries:
        handler = pack_handler(holder, entry)
        if decorate is not None:
            handler = decorate(handler)
        registry.register(
            f"{NAME_PREFIX}{entry.key}",
            entry.triggers,
            handler,
            description_for(entry),
            groups=pack.groups,
            takes_args=True,
        )