signal-bot/signal_bot/pack.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""A private album, and the shape of the commands that post from it.

Nothing here knows a single name. Who the people are, what somebody types to
get their photo, and which file shows whom are all read at run time from a
directory outside this repository -- so the mechanism can be public, tested
and committed while the album stays private. An installation without a pack
directory simply has no such commands.

The pack refers to people by opaque keys (``p1``, ``p2``). Those keys, never
the triggers, are what reach a log line or the rate-limit table: a name that
never leaves that directory cannot leak out of one.

The entry shape is deliberately one type rather than three. A person is an
entry with a single option; "a photo with both of them" is an entry with one
two-key option; "either of them" is an entry with two one-key options. The
handler then has exactly one job -- pick an option that actually has photos
-- and the difference between the three cases is data, not code.

Loading and validating a pack is pack_load.py, the way config_env.py is to
config.py: this half is the shape, that half is how it is read.
"""

from __future__ import annotations

import dataclasses
import itertools
import os
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Final

from signal_bot.folding import fold
from signal_bot.pack_custom import CustomTags
from signal_bot.state_dir import state_dir

if TYPE_CHECKING:
    from collections.abc import Mapping, Sequence

PEOPLE_FILE: Final = "people.json"
TAGS_FILE: Final = "tags.json"
MEDIA_DIR: Final = "media"

# What Signal is told an attachment is. Closed, and checked at load time: the
# suffix is the only thing that tells Signal how to render the file, so an
# unknown one is refused rather than guessed. Mirrors commons.SUFFIXES.
MEDIA_TYPES: Final[dict[str, str]] = {
    ".jpg": "image/jpeg",
    ".jpeg": "image/jpeg",
    ".png": "image/png",
    ".gif": "image/gif",
    ".mp4": "video/mp4",
    # Everything !add_photo can ingest has to be here too, or a file that
    # uploaded cleanly is refused by the *next* load and takes startup down
    # with it. Kept in step with media_sniff.KINDS, which the tests assert.
    ".webp": "image/webp",
    ".mov": "video/quicktime",
    ".m4v": "video/x-m4v",
    ".heic": "image/heic",
    ".heif": "image/heif",
    ".avif": "image/avif",
    ".webm": "video/webm",
    ".svg": "image/svg+xml",
}

# How many names one message may combine, the head included. A ceiling
# rather than a judgement: the option sets multiply, and nobody is asking
# for a photo of nine people at once.
MAX_NAMES: Final = 6


@dataclass(frozen=True, slots=True)
class Entry:
    """One command the pack defines, and the people it stands for."""

    # The pseudonym. The only identifier this program logs or stores.
    key: str
    # Raw triggers, canonical spelling first -- that is what !pomoc prints.
    triggers: tuple[str, ...]
    # Each element is one set of person keys that satisfies the request. One
    # option means there is nothing to choose; several mean the handler picks
    # between them, preferring one that has photos.
    options: tuple[frozenset[str], ...]


@dataclass(frozen=True, slots=True)
class Pack:
    """A loaded album: its commands, its tags, and where its files are."""

    directory: Path
    # The only groups these commands exist in. Everywhere else they are not
    # commands at all -- not refused, not answered, not logged.
    groups: frozenset[str]
    entries: tuple[Entry, ...]
    # Filename to the person keys tagged in it.
    tags: Mapping[str, frozenset[str]]
    # Folded trigger to entry, so a tail token resolves exactly the way the
    # command in front of it did.
    by_trigger: Mapping[str, Entry]
    # Labels !add_photo minted that match no person. These define no
    # commands -- !random is the only way to one -- which is what keeps a
    # hot-swap a data swap rather than a re-registration.
    custom: CustomTags = dataclasses.field(default_factory=CustomTags)

    def media_path(self, filename: str) -> Path:
        """Where one tagged file lives."""
        return self.directory / MEDIA_DIR / filename

    def matching(self, keys: frozenset[str]) -> tuple[str, ...]:
        """Every file tagged with all of ``keys``, in a stable order."""
        return tuple(sorted(name for name, tagged in self.tags.items() if keys <= tagged))


def media_type(filename: str) -> str:
    """The MIME type for a tagged file, by suffix."""
    return MEDIA_TYPES[Path(filename).suffix.lower()]


def default_dirs() -> tuple[tuple[Path, bool], ...]:
    """Every album directory, and whether the operator named it explicitly.

    A list, because one bot can serve several albums -- each scoped to its
    own groups, which is what lets a test group have its own invented people
    without the real album being readable there. PHOTO_PACK_DIR takes a
    comma-separated list, the same spelling GROUP_IDS and TRUSTED_INVITERS
    already use.
    """
    explicit = os.environ.get("PHOTO_PACK_DIR", "").strip()
    if explicit:
        return tuple((Path(part.strip()), True) for part in explicit.split(",") if part.strip())
    return (_implicit_dir(),)


def _implicit_dir() -> tuple[Path, bool]:
    """Where the pack lives when nobody said, and whether that was explicit.

    Explicit means a missing directory is an error worth failing startup for.
    The implicit paths are the opposite: they are where a pack *would* live,
    so their absence is simply an installation with no album.
    """
    return state_dir() / "pack", False


def options_for(pack: Pack, token: str) -> tuple[frozenset[str], ...] | None:
    """What one typed name could mean, or None when it means nothing.

    Accepts the bare spelling as well as the "!" one, because "!celina basia"
    is how people actually write it, and refusing the second half over a
    missing punctuation mark would be a rule nobody can see.
    """
    folded = fold(token)
    entry = pack.by_trigger.get(folded) or pack.by_trigger.get(f"!{folded}")
    return entry.options if entry is not None else None


def candidate_keysets(
    pack: Pack, entry: Entry, tail: Sequence[str]
) -> tuple[frozenset[str], ...] | None:
    """Every set of people the request could be asking for.

    None when the message names more people than MAX_NAMES, or when a word in
    the tail is not a name at all -- which is answered rather than ignored:
    silently dropping it would post a photo of the wrong subset and look
    exactly like the tags being wrong.
    """
    if len(tail) >= MAX_NAMES:
        return None
    choices = [entry.options]
    for token in tail:
        options = options_for(pack, token)
        if options is None:
            return None
        choices.append(options)
    return tuple(frozenset().union(*combination) for combination in itertools.product(*choices))