signal-bot/signal_bot/pack_custom.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Custom tags: labels the album learned from the chat rather than a manifest.

``!add_photo`` never writes a person. A label that matches nobody becomes a
custom tag instead -- a pseudonym key (``t1``, ``t2``) plus the raw spellings
that reach it -- and the only way to that key is ``!random <label>``. Keeping
people and custom tags in different files is what makes that guarantee cheap:
``people.json`` is edited by the tagger and defines commands, ``custom.json``
is appended to from the chat and defines no commands at all.

Written atomically, temp-then-replace, exactly as ``pack_album.Album.write``
already does, because the alternative is a half-written manifest that fails the
next load and takes startup down.

This module also owns the validation ``pack_load`` would otherwise grow: that
file was at 243 of the 250-line cap before this feature existed.
"""

from __future__ import annotations

import json
import re
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, Final

from signal_bot.folding import fold
from signal_bot.groups import ConfigError

if TYPE_CHECKING:
    from collections.abc import Mapping
    from pathlib import Path

CUSTOM_FILE: Final = "custom.json"

# The pseudonym shape. "t" for tag, mirroring "p" for person, and a digit
# suffix so a key can never be mistaken for something somebody typed.
KEY_PATTERN: Final = re.compile(r"^t([1-9][0-9]*)$")

# Manifests are written 0600: the album is 0700 already, but a mode is
# cheaper to keep right than to explain.
FILE_MODE: Final = 0o600


@dataclass(frozen=True, slots=True)
class CustomTags:
    """Every custom tag the album knows, in both directions."""

    # Key to its raw spellings, canonical first. The spellings are the only
    # place a typed label is kept, and that file never leaves the album.
    by_key: Mapping[str, tuple[str, ...]] = field(default_factory=dict)
    # Folded label to key, which is how a typed word becomes a pseudonym.
    by_label: Mapping[str, str] = field(default_factory=dict)

    @property
    def keys(self) -> frozenset[str]:
        """Every custom key, for validating tags.json against."""
        return frozenset(self.by_key)

    def resolve(self, label: str) -> str | None:
        """The key one typed label means, or None when nothing does."""
        return self.by_label.get(fold(label))

    def next_key(self) -> str:
        """The key a new label would mint, without minting it.

        One past the highest number in use rather than the count, so
        deleting ``t2`` by hand can never make a later mint reuse it and
        silently inherit its photos.
        """
        used = [int(match.group(1)) for key in self.by_key if (match := KEY_PATTERN.match(key))]
        return f"t{max(used, default=0) + 1}"

    def with_label(self, key: str, label: str) -> CustomTags:
        """This table plus one more label, as a new value."""
        by_key = dict(self.by_key)
        by_key[key] = (*by_key.get(key, ()), label)
        by_label = dict(self.by_label)
        by_label[fold(label)] = key
        return CustomTags(by_key=by_key, by_label=by_label)


def load(directory: Path, person_triggers: frozenset[str]) -> CustomTags:
    """Read custom.json, or return an empty table when there is none.

    A missing file is the normal state of an album nobody has run
    ``!add_photo`` on, so it is not an error. A malformed one is: it would
    otherwise drop every custom tag silently and make ``!random`` answer "no
    such photo" for labels that are right there on disk.

    Args:
        directory: The pack directory.
        person_triggers: Folded triggers already claimed by people, so a
            label can never shadow a person's own command.

    Raises:
        ConfigError: The file is unreadable, malformed, or ambiguous.
    """
    path = directory / CUSTOM_FILE
    if not path.exists():
        return CustomTags()
    document = _document(path)
    by_key: dict[str, tuple[str, ...]] = {}
    by_label: dict[str, str] = {}
    for raw_key, raw_labels in document.items():
        key = _key(str(raw_key))
        labels = _labels(key, raw_labels)
        by_key[key] = labels
        for label in labels:
            _claim(by_label, fold(label), key, person_triggers)
    return CustomTags(by_key=by_key, by_label=by_label)


def write(directory: Path, tags: CustomTags) -> None:
    """Replace custom.json atomically, so a crash cannot halve it."""
    path = directory / CUSTOM_FILE
    temporary = path.with_suffix(".tmp")
    temporary.write_text(
        json.dumps({key: list(labels) for key, labels in tags.by_key.items()}, indent=2),
        encoding="utf-8",
    )
    temporary.chmod(FILE_MODE)
    temporary.replace(path)


def _document(path: Path) -> dict[str, Any]:
    """One JSON object, read strictly."""
    try:
        text = path.read_text(encoding="utf-8")
    except OSError as error:
        msg = f"{CUSTOM_FILE} cannot be read: {error}"
        raise ConfigError(msg) from error
    try:
        document = json.loads(text)
    except json.JSONDecodeError as error:
        msg = f"{CUSTOM_FILE} is not valid JSON: {error}"
        raise ConfigError(msg) from error
    if not isinstance(document, dict):
        msg = f"{CUSTOM_FILE} must be an object"
        raise ConfigError(msg)
    return document


def _key(key: str) -> str:
    """One custom key, which has to look like a pseudonym and not a name."""
    if not KEY_PATTERN.match(key):
        msg = f"{CUSTOM_FILE} key {key!r} must be 't' followed by a number"
        raise ConfigError(msg)
    return key


def _labels(key: str, raw: object) -> tuple[str, ...]:
    """The spellings one custom key answers to."""
    if not isinstance(raw, list) or not raw:
        msg = f"{CUSTOM_FILE} entry {key!r} needs a non-empty list of labels"
        raise ConfigError(msg)
    labels = tuple(str(label) for label in raw)
    for label in labels:
        if not fold(label).strip("!"):
            msg = f"{CUSTOM_FILE} label {label!r} folds away to nothing"
            raise ConfigError(msg)
    return labels


def _claim(
    by_label: dict[str, str],
    folded: str,
    key: str,
    person_triggers: frozenset[str],
) -> None:
    """Record one label, refusing a spelling that already means something.

    Both collisions matter and neither is recoverable at run time: two custom
    keys sharing a label would make ``!random`` pick one arbitrarily, and a
    label that folds onto a person's trigger would quietly shadow the command
    that person's photos are reached by.
    """
    if folded in person_triggers or f"!{folded}" in person_triggers:
        msg = f"{CUSTOM_FILE} label {folded!r} is already a person's trigger"
        raise ConfigError(msg)
    owner = by_label.setdefault(folded, key)
    if owner != key:
        msg = f"{CUSTOM_FILE} label {folded!r} is claimed by {owner!r} and {key!r}"
        raise ConfigError(msg)