signal-bot/signal_bot/pack_load.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Reading a photo pack off disk, and refusing one it cannot serve.

Every check here fails startup rather than a command, for the reason
config_env.py gives: a manifest that names a person who does not exist, a
trigger the folder can never match, or a file that has since been renamed is
an answer, not an outage, and no amount of waiting fixes it. The one thing
that is *not* an error is a pack that has not been tagged yet -- that is an
album waiting for the tagger, and it says so in the log.
"""

from __future__ import annotations

import json
import logging
from typing import TYPE_CHECKING, Any, Final

from signal_bot.folding import fold
from signal_bot.groups import ConfigError
from signal_bot.pack import (
    MEDIA_DIR,
    MEDIA_TYPES,
    PEOPLE_FILE,
    TAGS_FILE,
    Entry,
    Pack,
)
from signal_bot.pack_custom import load as load_custom

if TYPE_CHECKING:
    from pathlib import Path

logger = logging.getLogger(__name__)

# "!" plus at least one character to fold.
MIN_TRIGGER_LENGTH: Final = 2


def load(directory: Path) -> Pack:
    """Read a pack, or say precisely what is wrong with it.

    Raises:
        ConfigError: The manifest is missing, malformed, or names a person, a
            trigger or a file the pack cannot honour.
    """
    document = _document(directory / PEOPLE_FILE)
    entries = _entries(document)
    known = {entry.key for entry in entries}
    named = {key for entry in entries for option in entry.options for key in option}
    unknown = named - known
    if unknown:
        msg = f"{PEOPLE_FILE} refers to unknown people: {sorted(unknown)}"
        raise ConfigError(msg)
    by_trigger = {fold(t): entry for entry in entries for t in entry.triggers}
    # Custom tags are keys tags.json may legitimately name, so they have to
    # be known before it is validated -- and they are checked against the
    # people's triggers so a label can never shadow somebody's command.
    custom = load_custom(directory, frozenset(by_trigger))
    return Pack(
        directory=directory,
        groups=_groups(document),
        entries=entries,
        tags=_tags(directory, named | custom.keys),
        by_trigger=by_trigger,
        custom=custom,
    )


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


def _groups(document: dict[str, Any]) -> frozenset[str]:
    """The groups the album may be posted into, which may not be empty.

    Fail-closed on purpose: the failure mode of defaulting to "every group
    the bot is in" is somebody's private photos in a stranger's chat.
    """
    groups = document.get("groups")
    if not isinstance(groups, list) or not groups:
        msg = f"{PEOPLE_FILE} needs a non-empty 'groups' list"
        raise ConfigError(msg)
    named = {str(group).strip() for group in groups}
    if "" in named:
        msg = f"{PEOPLE_FILE} has a blank group id"
        raise ConfigError(msg)
    return frozenset(named)


def _entries(document: dict[str, Any]) -> tuple[Entry, ...]:
    """One entry per person and per combination, validated as it is built."""
    entries = [
        Entry(
            key=_key(person, "people"),
            triggers=_triggers(person, "people"),
            options=(frozenset({_key(person, "people")}),),
        )
        for person in _listing(document, "people")
    ]
    entries.extend(
        Entry(
            key=_key(combo, "combos"), triggers=_triggers(combo, "combos"), options=_options(combo)
        )
        for combo in _listing(document, "combos")
    )
    if not entries:
        msg = f"{PEOPLE_FILE} defines no commands"
        raise ConfigError(msg)
    _unique_keys(entries)
    _unique_triggers(entries)
    return tuple(entries)


def _listing(document: dict[str, Any], field: str) -> list[dict[str, Any]]:
    """One list of manifest objects, defaulting to none at all."""
    entries = document.get(field, [])
    if not isinstance(entries, list) or any(not isinstance(entry, dict) for entry in entries):
        msg = f"{PEOPLE_FILE} field '{field}' must be a list of objects"
        raise ConfigError(msg)
    return entries


def _key(entry: dict[str, Any], field: str) -> str:
    """The pseudonym for one entry.

    Rejected when it looks like a name rather than a key, because this is the
    string that ends up in the log and in command_uses: the whole point is
    that reading either tells you nothing about who was asked for.
    """
    key = str(entry.get("key", "")).strip()
    if not key or not key.replace("_", "").isalnum() or not key.isascii():
        msg = f"{PEOPLE_FILE} {field} entry needs an ascii alphanumeric 'key', got {key!r}"
        raise ConfigError(msg)
    return key


def _triggers(entry: dict[str, Any], field: str) -> tuple[str, ...]:
    """Every spelling that fires one entry, canonical first."""
    triggers = entry.get("triggers")
    if not isinstance(triggers, list) or not triggers:
        msg = f"{PEOPLE_FILE} {field} entry {entry.get('key')!r} needs a non-empty 'triggers' list"
        raise ConfigError(msg)
    return tuple(_trigger(str(trigger), field) for trigger in triggers)


def _trigger(trigger: str, field: str) -> str:
    """One trigger, checked against what the registry can actually match."""
    reason = ""
    if not trigger.startswith("!") or len(trigger) < MIN_TRIGGER_LENGTH:
        reason = 'must start with "!" and have something after it'
    elif any(character.isdigit() for character in trigger):
        # folding maps 0134578 onto letters, so "!7dni" would resolve as
        # "!tdni" -- a trigger with a digit is a trigger you cannot type.
        reason = "must not contain a digit, which folding turns into a letter"
    elif not fold(trigger).strip("!"):
        reason = "folds away to nothing, so no message could ever match it"
    if reason:
        msg = f"{PEOPLE_FILE} {field} trigger {trigger!r} {reason}"
        raise ConfigError(msg)
    return trigger


def _options(combo: dict[str, Any]) -> tuple[frozenset[str], ...]:
    """What a combination stands for: everyone at once, or any one of them."""
    every = combo.get("all_of")
    either = combo.get("any_of")
    if isinstance(every, list) and every and either is None:
        return (frozenset(str(key) for key in every),)
    if isinstance(either, list) and either and every is None:
        return tuple(frozenset({str(key)}) for key in either)
    msg = f"{PEOPLE_FILE} combo {combo.get('key')!r} needs exactly one of 'all_of' or 'any_of'"
    raise ConfigError(msg)


def _unique_keys(entries: list[Entry]) -> None:
    """Two entries sharing a key would share a cooldown and a log line."""
    seen: set[str] = set()
    for entry in entries:
        if entry.key in seen:
            msg = f"{PEOPLE_FILE} uses the key {entry.key!r} twice"
            raise ConfigError(msg)
        seen.add(entry.key)


def _unique_triggers(entries: list[Entry]) -> None:
    """Caught here rather than by the registry, which cannot name the file.

    Skeletons, not spellings: two triggers can look different and fold the
    same, because folding strips diacritics. Within one entry that is normal
    and wanted -- "!celina" and "!celiną" are one person written two ways,
    and the registry collapses them into one frozenset. Across two entries it
    is a manifest that cannot work, and the registry would only be able to
    say which skeleton collided, not which entry to fix.
    """
    claimed: dict[str, str] = {}
    for entry in entries:
        for trigger in entry.triggers:
            owner = claimed.setdefault(fold(trigger), entry.key)
            if owner != entry.key:
                msg = f"{PEOPLE_FILE} trigger {trigger!r} is claimed by {owner!r} and {entry.key!r}"
                raise ConfigError(msg)


def _tags(directory: Path, people: set[str]) -> dict[str, frozenset[str]]:
    """Which file shows whom, checked against the album and the manifest.

    A missing tags.json is not an error: that is a pack the tagger has not
    been run on yet, and the commands answer "no such photo" until it has.
    A tag naming a file that is not there *is* an error -- a renamed file
    would otherwise drop out of the album silently.
    """
    path = directory / TAGS_FILE
    if not path.exists():
        logger.warning("photo pack %s has no %s yet; run tool/tagger.py", directory, TAGS_FILE)
        return {}
    document = _document(path)
    tags: dict[str, frozenset[str]] = {}
    for filename, keys in document.items():
        _media_file(directory, str(filename))
        if not isinstance(keys, list) or any(str(key) not in people for key in keys):
            msg = f"{TAGS_FILE} entry {filename!r} names somebody who is not in {PEOPLE_FILE}"
            raise ConfigError(msg)
        if keys:
            tags[str(filename)] = frozenset(str(key) for key in keys)
    return tags


def _media_file(directory: Path, filename: str) -> None:
    """The tagged file has to exist, and be something Signal can render."""
    path = directory / MEDIA_DIR / filename
    if path.name != filename or not path.is_file():
        msg = f"{TAGS_FILE} names {filename!r}, which is not a file in {MEDIA_DIR}/"
        raise ConfigError(msg)
    if path.suffix.lower() not in MEDIA_TYPES:
        msg = f"{TAGS_FILE} names {filename!r}, whose type is not one Signal is told about"
        raise ConfigError(msg)