kuhyx/signal-bot
1
0
Fork 0
mirror of https://github.com/kuhyx/signal-bot.git synced 2026-11-11 15:54:21 +01:00
Code Issues Actions Packages Projects Releases Wiki Activity

release: 0.2.0 «Urbanus I» — a private photo album, and a tagger for it

Browse Source 
Per-person photo commands, where the mechanism is public and the album is
not. Nothing identifying is in this repository: who the people are, what
fires their photos and which file shows whom all live in one directory
outside it, and the tests, the docs and the offline e2e run use three
invented people instead.

Several names in one message intersect -- "!alfa beta" is a photo with both
-- and where an entry offers a choice the bot picks among the readings that
actually have a photo, so a question the album could answer is not refused.
A combination nobody photographed gets one line rather than silence, which
from the group's side is indistinguishable from a broken bot.

Scoping is enforced in the dispatcher, not in the handler: a command the
group is not allowed resolves to nothing, so it is not answered, not counted
as a command, and never named in a log line outside its own groups. A
handler that quietly returned would leak all three. Commands are named by
pseudonym (pack:p1), so the journal and command_uses say somebody asked for
a photo and nothing about whose.

Video works: the attachment goes out as data:video/mp4;filename=...;base64,
which is the container's own spelling, read off its /swagger/doc.json.
Told it is a photo, Signal renders a clip as an unplayable file.

tool/tagger.py serves one page on loopback to fill the album in -- names are
typed into the page, never as arguments, and it prints how many photos each
command would have when it stops, because a command with zero is dead on
arrival.

Also renames the real people who were still sitting in the stats test
fixtures. They remain in the history; the point is that nothing new is
added to it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017w1HZ3cZ7neL3rFdQ6VMBA
This commit is contained in:
Krzysztof Rudnicki 2026-08-30 22:35:41 +02:00
parent 116e229488
commit d00a646f17
46 changed files with 2439 additions and 95 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
.env.sample
View File

@ -71,6 +71,17 @@ PORTRAIT_CACHE_DIR=
# command without a second account, and it posts for real.
SELFTEST_GROUPS=
# Optional, and private. The album the per-person photo commands draw
# from: a directory holding people.json, tags.json and media/. Everything
# identifying lives in there and nothing about it is in this repository --
# see DOCS-photo-pack.md, and tool/tagger.py for how it gets filled in.
#
# Leave it empty and the bot looks in $STATE_DIRECTORY/pack (systemd) and
# then $XDG_DATA_HOME/signal-bot/pack. Not finding one is normal: the
# commands simply do not exist. Naming one that is not there, or one whose
# manifest cannot be honoured, fails startup rather than every command.
PHOTO_PACK_DIR=
# Optional. Where the /healthz readiness probe binds.
# Default: 127.0.0.1:8137 -- deliberately not 8000, which unrelated dev
# servers grab, and below the ephemeral range so the kernel cannot take it.

13
.gitignore vendored
View File

@ -164,9 +164,22 @@ cython_debug/
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/
*.jpg
*.jpeg
*.png
*.webp
*.gif
*.mp4
nohup.out
# The private photo album. It lives outside this repository -- in
# ~/.local/share/signal-bot/pack -- and these names exist so that a stray
# copy inside it cannot be committed by accident. people.json and tags.json
# are the two files that hold the names; media/ holds the photos.
people.json
tags.json
media/
pack/
# The message database is private and never leaves this machine. A live
# SQLite database is three files, so all three are named.
*.sqlite3

14
DOCS-changelog.md
View File

@ -71,6 +71,20 @@ this repo: describe what changes for them. 0.0.1 shipped a bullet about
`/healthz`, an endpoint bound to 127.0.0.1 that nobody in a Signal group can
reach, and it was rightly called out.
## 0.2.0 — Urbanus I
papież nr 17 z 267, pontyfikat 222-230
- nowe komendy ze zdjęciami: każda osoba ma swoją, a kilka naraz
(`!ktos !ktos2`) daje zdjęcie, na którym są wszyscy naraz
- jak takiego zdjęcia nie ma, bot mówi to jednym zdaniem zamiast milczeć
- filmiki też działają — lecą jako filmiki, a nie jako plik, którego nie da
się odtworzyć
- te komendy istnieją tylko w tej jednej grupie: w każdej innej nie ma ich
wcale, `!pomoc` ich tam nie pokazuje i bot na nie nie odpowiada
- zdjęcia i imiona nie są w kodzie bota ani nigdzie w internecie — leżą w
jednym katalogu na komputerze i nigdy go nie opuszczają
## 0.1.7 — Callistus I
papież nr 16 z 267, pontyfikat 217-222

107
DOCS-photo-pack.md Normal file
View File

@ -0,0 +1,107 @@
# The photo pack
A pack is a private album the bot posts from: `!alfa` returns a random photo
of alfa, and `!alfa beta` returns one with both of them on it.
**Nothing about a real pack is in this repository, and nothing ever will be.**
Who the people are, what somebody types to get their photo and which file
shows whom live in one directory outside the repo. This document, the tests
and the offline end-to-end run all use the same invented people — alfa, beta,
gamma — so the mechanism can be public while the album stays private.
## Where it lives
```
~/.local/share/signal-bot/pack/
people.json # the commands, and the only place names exist
tags.json # which file shows whom
media/ # the photos and videos themselves
```
Found in this order: `PHOTO_PACK_DIR`, then `$STATE_DIRECTORY/pack` (what
systemd hands a **system** unit), then `$XDG_DATA_HOME/signal-bot/pack`.
Under the *user* unit this repo installs, `StateDirectory=` would resolve
below `$XDG_STATE_HOME` rather than `$XDG_DATA_HOME`, so the deployed bot
names its pack outright in `.env` instead of relying on discovery.
Two absences, deliberately different:
- a pack **nobody named** that is not there, or that has no `people.json`
yet, is an installation without an album: the commands do not exist, and
that is the normal case on every machine and in the whole test suite;
- a pack named in `PHOTO_PACK_DIR` that is missing, or whose manifest cannot
be honoured, **fails startup**. `load_config` raising is an answer, and a
bot that came up healthy and said "no such photo" forever would not be.
## people.json
```json
{
"groups": ["group.WlRKbExYZGhkR05vWldRdA=="],
"people": [
{"key": "p1", "triggers": ["!alfa", "!alfka"]},
{"key": "p2", "triggers": ["!beta"]}
],
"combos": [
{"key": "c1", "triggers": ["!oboje"], "all_of": ["p1", "p2"]},
{"key": "c2", "triggers": ["!ktokolwiek"], "any_of": ["p1", "p2"]}
]
}
```
- **`groups` is the whole privacy guarantee.** In any other group these are
not commands at all: not answered, not counted as a command, not named in a
log line. A handler that quietly returned would still leak all three, which
is why the *dispatcher* enforces it — `registry.resolve_with_args` takes the
group and gives back nothing outside it.
- **`key` is a pseudonym**, and the only identifier that reaches the journal
or the `command_uses` table. A command is `pack:p1`; reading either says
somebody asked for a photo and nothing about whose.
- **`all_of` means one photo with all of them; `any_of` means either**, and
where a choice exists the bot picks among the readings that actually have a
photo, so a combination that could be answered is not refused.
- Triggers may not contain a digit — `folding.py` maps `0134578` onto letters,
so `!alfa7` is not what you typed. Two spellings that fold the same are fine
*within one entry* (`!ala`/`!ala` is one person written twice) and
refused across two.
## tags.json
```json
{"alfa.png": ["p1"], "both.png": ["p1", "p2"], "clip.mp4": ["p1"]}
```
A tag naming a file that is not in `media/` fails startup: a renamed file
would otherwise drop out of the album in silence. An empty list means "seen,
nobody in it", which is how the tagger records a photo you skipped.
Videos work. The attachment goes out as
`data:video/mp4;filename=<name>;base64,<data>` — the spelling is the
container's own, from its `/swagger/doc.json` — because told it is a photo,
Signal renders a clip as a file nobody can play.
## Filling it in: tool/tagger.py
```sh
python tool/tagger.py --import ~/photos # copy the files in, then tag
python tool/tagger.py # carry on where you left off
```
It serves one page on `127.0.0.1:8140`: the item, a numbered button per
person, `1`–`9` to toggle, `→`/`←` to move. Every toggle rewrites `tags.json`
atomically, so closing the tab loses nothing. Names are typed **into the
page**, never as arguments — an argument would land in the shell history —
and the group is picked from the live `/v1/groups` listing rather than pasted.
When you stop it, it prints how many items were never opened and how many
photos each command would have. **A command showing `0` is dead on arrival**;
that is the number to check before restarting the bot, not after somebody in
the group tries it.
## Running the mechanism without an album
`scripts/e2e.sh` writes a synthetic pack with `tool/e2e_pack.py` and proves,
offline and with no Signal account, that one name posts a photo, two names
intersect, an impossible combination gets one line and no attachment, a video
leaves as a video, and the same command in another group produces **zero**
sends.

19
fake_signal/scenario.py
View File

@ -5,7 +5,7 @@ from __future__ import annotations
import json
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING, Any, Final
import yaml
@ -149,9 +149,22 @@ def load_scenario(path: Path) -> Scenario:
)
# How signal-cli-rest-api spells an attachment whose type the bot stated.
# Anything else is bare base64, which is what an ordinary photo travels as.
_VIDEO_PREFIX: Final = "data:video/"
def describe(payload: dict[str, Any]) -> tuple[str, str, str]:
"""Reduce a captured send to (kind, recipient, body) for comparison."""
kind = "image" if payload.get("base64_attachments") else "text"
"""Reduce a captured send to (kind, recipient, body) for comparison.
"video" is its own kind so a scenario can assert that a clip went out as
a clip: told it is a photo, Signal renders an unplayable file, and every
other check in this harness would still pass.
"""
attachments = payload.get("base64_attachments") or []
kind = "text"
if attachments:
kind = "video" if str(attachments[0]).startswith(_VIDEO_PREFIX) else "image"
recipients = payload.get("recipients") or [""]
return kind, str(recipients[0]), str(payload.get("message", ""))

7
fake_signal/scenarios/pack_combination.yaml Normal file
View File

@ -0,0 +1,7 @@
# The whole point of the feature: several names mean "a photo with all of
# them", and only both.png is tagged with p1 and p2 at once.
name: "two names ask for a photo with both people on it"
inject:
- { text: "!alfa beta", uuid: uuid-pack-two, name: Pack2 }
expect:
- { kind: image, to: reply }

7
fake_signal/scenarios/pack_command.yaml Normal file
View File

@ -0,0 +1,7 @@
# The private album, against a synthetic pack the e2e script writes: the real
# one lives outside this repository and must never be needed to run the suite.
name: "!alfa posts a photo from the private album"
inject:
- { text: "!alfa", uuid: uuid-pack-one, name: Pack1 }
expect:
- { kind: image, to: reply }

7
fake_signal/scenarios/pack_no_photo.yaml Normal file
View File

@ -0,0 +1,7 @@
# A combination nobody photographed is answered, not met with silence: from
# the group's side, silence is indistinguishable from a broken bot.
name: "a combination with no photo gets one line and no attachment"
inject:
- { text: "!beta gamma", uuid: uuid-pack-three, name: Pack3 }
expect:
- { kind: text, to: reply, contains: "NIE MAM TAKIEGO" }

7
fake_signal/scenarios/pack_scoping.yaml Normal file
View File

@ -0,0 +1,7 @@
# The privacy guarantee. In a group the pack does not list, "!alfa" is not a
# command at all: no photo, no refusal, nothing -- a refusal would itself
# announce that the album exists.
name: "the album does not exist in a group it was not given to"
inject:
- { text: "!alfa", group: second, uuid: uuid-pack-five, name: Pack5 }
expect: []

7
fake_signal/scenarios/pack_video.yaml Normal file
View File

@ -0,0 +1,7 @@
# Told it is a photo, Signal renders a clip as a file nobody can play. The
# harness reads the data URI the bot sent, so this fails if the type is lost.
name: "a tagged video leaves as a video, not as a photo"
inject:
- { text: "!wideo", uuid: uuid-pack-four, name: Pack4 }
expect:
- { kind: video, to: reply }

2
pyproject.toml
View File

@ -1,6 +1,6 @@
[project]
name = "signal-bot"
version = "0.1.7"
version = "0.2.0"
description = "A Signal group bot: image commands and daily message tallies."
readme = "README.md"
license = "MIT"

8
scripts/e2e.sh
View File

@ -120,6 +120,13 @@ main() {
# would otherwise let the developer's own .env decide what this run
# watches -- and a TRUSTED_INVITERS there would refuse every fixture
# group. The gate itself is covered in tests/test_trust.py.
# The album is private and lives outside this repository, so the run
# builds a synthetic one: three invented people, a combination and a
# video, scoped to the watched group so the scoping scenario can prove
# it does not exist in the second one.
echo "== writing a synthetic photo pack"
"$PYTHON" "$REPO_ROOT/tool/e2e_pack.py" --dir "${TEMP_DIR}/pack" --group "$WATCHED_GROUP"
echo "== booting the bot on :${BOT_PORT}, pointed at the fake"
env -u CAT_API \
PHONE_NUMBER="+15550000000" \
@ -132,6 +139,7 @@ main() {
GECKO_API_URL="${FAKE_URL}/_fixtures/gecko-api" \
COMMONS_API_URL="${FAKE_URL}/_fixtures/commons-api" \
PORTRAIT_CACHE_DIR="${TEMP_DIR}/portraits" \
PHOTO_PACK_DIR="${TEMP_DIR}/pack" \
BIND_PORT="$BOT_PORT" \
SILENCE_SECONDS=0 \
"$PYTHON" -m signal_bot &

7
signal_bot/bot.py
View File

@ -21,6 +21,8 @@ from signal_bot.decorators import cooldown
from signal_bot.dispatcher import Dispatcher
from signal_bot.help_command import register_help_command
from signal_bot.http_client import ImageFetcher
from signal_bot.pack_command import register_pack_commands
from signal_bot.pack_load import load as load_pack
from signal_bot.popes_command import register_pope_command
from signal_bot.portraits import PortraitCache
from signal_bot.registry import CommandRegistry
@ -220,6 +222,11 @@ def build_bot(
)
register_pope_command(registry, portraits, decorate=rate_limit)
register_stats_commands(registry, decorate=rate_limit)
# Absent unless this installation has an album, and scoped to that
# album's own groups when it does -- so on every other machine, and in
# every other group, these commands simply do not exist.
if config.photo_pack_dir is not None:
register_pack_commands(registry, load_pack(config.photo_pack_dir), decorate=rate_limit)
# Last, so !pomoc trails the list it prints; it reads the registry
# when it runs, so the order is presentation only.
register_help_command(registry, decorate=rate_limit)

9
signal_bot/config.py
View File

@ -4,9 +4,12 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Final
from typing import TYPE_CHECKING, Final
from urllib.parse import urlsplit, urlunsplit
if TYPE_CHECKING:
from pathlib import Path
from signal_bot.endpoints import (
DEFAULT_CAT_API_URL,
DEFAULT_COMMONS_API_URL,
@ -93,6 +96,10 @@ class Config:
# Lowered in the e2e run so the deafness gate can be proven against the
# real process rather than only in unit tests.
silence_seconds: int = DEFAULT_SILENCE_SECONDS
# The private photo album, or None for an installation without one. Only
# the path is config; everything inside it -- who the people are, what
# fires their photos -- is data this repository never sees.
photo_pack_dir: Path | None = None
@property
def is_pinned(self) -> bool:

35
signal_bot/config_env.py
View File

@ -41,10 +41,13 @@ from signal_bot.endpoints import (
DEFAULT_GECKO_API_URL,
)
from signal_bot.groups import ConfigError, parse_group_ids
from signal_bot.pack import PEOPLE_FILE, default_dir
from signal_bot.pack_load import load as load_pack
from signal_bot.trust import parse_trusted_inviters
if TYPE_CHECKING:
from collections.abc import Mapping
from pathlib import Path
__all__ = ["load_config"]
@ -102,6 +105,37 @@ def _bind_port(env: Mapping[str, str]) -> int:
return bind_port
def _photo_pack_dir() -> Path | None:
"""The album directory, loaded once here to prove it can be served.
Two different absences, on purpose. A directory the operator *named* and
that is not there is a typo, and startup says so. A directory nobody
named and that is not there is simply an installation with no album, and
the commands do not exist -- which is also what every test gets.
The pack is parsed and thrown away rather than trusted: a manifest that
cannot be honoured should fail here, where load_config raising is an
answer, and not later inside a retry loop that would hide it.
"""
directory, explicit = default_dir()
if not directory.is_dir():
if explicit:
msg = f"PHOTO_PACK_DIR={directory} is not a directory"
raise ConfigError(msg)
return None
if not (directory / PEOPLE_FILE).is_file():
# A directory the tagger has created but not filled in yet. Only an
# explicitly named pack may fail here: the implicit path is a place a
# pack *would* live, and failing on it would put the unit into a
# restart loop the moment somebody made the directory.
if explicit:
msg = f"PHOTO_PACK_DIR={directory} has no {PEOPLE_FILE}"
raise ConfigError(msg)
return None
load_pack(directory)
return directory
def load_config(env: dict[str, str] | None = None, *, use_dotenv: bool = True) -> Config:
"""Build a Config from the environment, failing loudly on anything missing.
@ -135,4 +169,5 @@ def load_config(env: dict[str, str] | None = None, *, use_dotenv: bool = True) -
log_level=_log_level(env),
trusted_inviters=parse_trusted_inviters(env),
silence_seconds=_positive_int(env, "SILENCE_SECONDS", DEFAULT_SILENCE_SECONDS),
photo_pack_dir=_photo_pack_dir(),
)

9
signal_bot/dispatcher.py
View File

@ -74,7 +74,13 @@ class Dispatcher:
# Resolved before recording, not after: is_command is part of the
# row, so "was this a command" has to be known at write time. It is
# the same lookup the handler below uses, not a second one.
command = self.registry.resolve(message.text)
#
# Scoped to this group, and that is the whole enforcement: a command
# a group is not allowed comes back as None here, so it is not
# answered, not recorded as a command and never named in a log line
# outside the groups it belongs to.
resolved = self.registry.resolve_with_args(message.text, group)
command = resolved.command if resolved is not None else None
if should_count(message):
record_message(
self.store.events,
@ -98,6 +104,7 @@ class Dispatcher:
command_name=command.name,
now=now.timestamp(),
day=now.date().isoformat(),
args=resolved.args if resolved is not None else (),
)
try:
await command.handler(ctx)

49
signal_bot/help_command.py
View File

@ -20,6 +20,7 @@ from __future__ import annotations
from typing import TYPE_CHECKING, Final
from signal_bot.animals import load_aliases
from signal_bot.pack_command import NAME_PREFIX
if TYPE_CHECKING:
from collections.abc import Sequence
@ -59,15 +60,44 @@ GROUPS: Final[tuple[tuple[str, tuple[str, ...]], ...]] = (
# list that says "!kot".
SPELLINGS: Final = "Każda komenda ma też inne pisownie, np. !cat, !stats, !week."
# The private album, when a group has one. Its heading is here and its
# commands are not: they come from a file this repository never sees, so
# GROUPS could not name them even if it wanted to -- and naming an optional
# command there would make !pomoc fail in every group that has no pack.
PACK_HEADING: Final = "Zdjęcia"
PACK_HINT: Final = "Kilka naraz ({first} {second}) = zdjęcie, na którym są wszyscy."
def _line(command: Command) -> str:
"""One command as !pomoc prints it."""
return f"{command.primary} — {command.description}"
def render_help(commands: Sequence[Command]) -> str:
"""Render the whole command list, grouped, in registration order."""
by_name = {command.name: command for command in commands}
def _pack_block(commands: Sequence[Command]) -> str:
"""The album's own block, or "" for a group that has no album.
Built from whatever is registered rather than from a table, because the
entries are private data: this file cannot know how many there are or
what they are called, and must not need to.
"""
entries = [c for c in commands if c.name.startswith(NAME_PREFIX)]
if not entries:
return ""
lines = [PACK_HEADING, *(_line(command) for command in entries)]
if len(entries) > 1:
lines.append(PACK_HINT.format(first=entries[0].primary, second=entries[1].primary))
return "\n".join(lines)
def render_help(commands: Sequence[Command], group_id: str) -> str:
"""Render the command list for one group, grouped, in registration order.
Per group, because a command can be scoped to one: the private album
exists in its own groups and nowhere else, and a help text that listed it
everywhere would announce it to every group the bot is in.
"""
visible = [command for command in commands if command.allowed_in(group_id)]
by_name = {command.name: command for command in visible}
# Indexed, not .get(): a name in GROUPS that no command answers to is a
# typo in this file, and it should fail loudly the first time the command
# is used rather than silently drop a line.
@ -75,7 +105,16 @@ def render_help(commands: Sequence[Command]) -> str:
blocks = [HEADING]
for title, names in GROUPS:
blocks.append("\n".join([title, *(_line(by_name[name]) for name in names)]))
blocks.append("\n".join(_line(c) for c in commands if c.name not in grouped))
pack = _pack_block(visible)
if pack:
blocks.append(pack)
blocks.append(
"\n".join(
_line(c)
for c in visible
if c.name not in grouped and not c.name.startswith(NAME_PREFIX)
)
)
blocks.append(SPELLINGS)
return "\n\n".join(blocks)
@ -94,7 +133,7 @@ def register_help_command(
table = load_aliases() if aliases is None else aliases
async def handler(ctx: CommandContext) -> None:
await ctx.reply(render_help(registry.commands))
await ctx.reply(render_help(registry.commands, ctx.group_id))
wrapped = decorate(handler) if decorate is not None else handler
registry.register(COMMAND_NAME, table[COMMAND_NAME], wrapped, DESCRIPTION)

155
signal_bot/pack.py Normal file
View File

@ -0,0 +1,155 @@
# 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 itertools
import os
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Final
from signal_bot.folding import fold
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 portraits._SUFFIXES.
MEDIA_TYPES: Final[dict[str, str]] = {
".jpg": "image/jpeg",
".jpeg": "image/jpeg",
".png": "image/png",
".gif": "image/gif",
".mp4": "video/mp4",
}
# 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]
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_dir() -> tuple[Path, bool]:
"""Where the pack lives, and whether the operator named it explicitly.
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.
Read straight off os.environ rather than from a passed mapping, exactly
as portraits.default_cache_dir is: .env has already been loaded into the
environment by then, and it is what lets one autouse fixture keep the
whole test suite away from a real album. STATE_DIRECTORY comes before
anything under $HOME because that is the directory systemd creates and
exposes regardless of ProtectHome.
"""
explicit = os.environ.get("PHOTO_PACK_DIR", "").strip()
if explicit:
return Path(explicit), True
state = os.environ.get("STATE_DIRECTORY", "").strip()
if state:
return Path(state) / "pack", False
home = os.environ.get("XDG_DATA_HOME", "").strip() or str(Path.home() / ".local" / "share")
return Path(home) / "signal-bot" / "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 marek"
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))

111
signal_bot/pack_command.py Normal file
View File

@ -0,0 +1,111 @@
# 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.pack import candidate_keysets, media_type
if TYPE_CHECKING:
from collections.abc import Sequence
from signal_bot.decorators import Decorator
from signal_bot.pack import Entry, Pack
from signal_bot.registry import CommandContext, 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."
# 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(pack: Pack, entry: Entry) -> Handler:
"""Build the handler that answers one entry's trigger."""
async def handler(ctx: CommandContext) -> None:
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)))
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,
pack: Pack,
*,
decorate: Decorator | None = None,
) -> None:
"""Register one command per pack entry, scoped to the pack's groups.
Args:
registry: Registry to populate.
pack: The loaded album.
decorate: Optional callable applied to each handler, used to attach
the cooldown gate without hard-wiring policy into the commands.
"""
for entry in pack.entries:
handler = pack_handler(pack, 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,
)

243
signal_bot/pack_load.py Normal file
View File

@ -0,0 +1,243 @@
# 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,
)
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)
return Pack(
directory=directory,
groups=_groups(document),
entries=entries,
tags=_tags(directory, named),
by_trigger={fold(t): entry for entry in entries for t in entry.triggers},
)
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 -- "!ala" and "!ala" 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)

75
signal_bot/registry.py
View File

@ -36,6 +36,9 @@ class CommandContext:
command_name: str
now: float
day: str
# Whatever followed the trigger, split on whitespace. Empty for every
# command that does not opt into taking any -- see Command.takes_args.
args: tuple[str, ...] = ()
async def reply(self, text: str) -> None:
"""Answer the message that triggered this command, in its group.
@ -53,6 +56,21 @@ class CommandContext:
image_base64, self.group_id, quote_of(self.message), caption
)
async def reply_media(self, payload_base64: str, media_type: str, filename: str) -> None:
"""Answer with an attachment that is not necessarily a photo.
Separate from reply_image because Signal decides how to render an
attachment from what it is told it is, and "a jpeg" is a lie that
makes a video arrive as a file nobody can play.
"""
await self.transport.send_media(
payload_base64,
self.group_id,
quote_of(self.message),
media_type=media_type,
filename=filename,
)
Handler = Callable[[CommandContext], Awaitable[None]]
@ -73,6 +91,20 @@ class Command:
# of "!papież". This is the first raw trigger instead, which is the
# canonical Polish one for every entry in aliases.json.
primary: str
# The only groups this command exists in, or None for everywhere. Not a
# gate inside the handler: the dispatcher treats a command that is not
# allowed here as no command at all, so it is never answered, never
# counted as a command and never logged by name in a group it does not
# belong to. A silent handler would still leak all three.
groups: frozenset[str] | None = None
# Whether words after the trigger are arguments rather than nonsense.
# Off by default, so "!cat blah" keeps resolving to nothing exactly as
# it does today.
takes_args: bool = False
def allowed_in(self, group_id: str) -> bool:
"""Whether this command exists in one group."""
return self.groups is None or group_id in self.groups
class DuplicateTriggerError(ValueError):
@ -89,6 +121,14 @@ class CommandRefusedError(Exception):
"""
@dataclass(frozen=True, slots=True)
class Resolved:
"""A command a message fired, and the words that followed it."""
command: Command
args: tuple[str, ...]
class CommandRegistry:
"""Maps trigger words to commands."""
@ -98,7 +138,14 @@ class CommandRegistry:
self._commands: list[Command] = []
def register(
self, name: str, triggers: Iterable[str], handler: Handler, description: str
self,
name: str,
triggers: Iterable[str],
handler: Handler,
description: str,
*,
groups: frozenset[str] | None = None,
takes_args: bool = False,
) -> Command:
"""Register one command under every one of its triggers.
@ -117,6 +164,8 @@ class CommandRegistry:
handler=handler,
description=description,
primary=raw[0],
groups=groups,
takes_args=takes_args,
)
for trigger in normalised:
self._by_trigger[trigger] = command
@ -134,6 +183,30 @@ class CommandRegistry:
return None
return self._by_trigger.get(candidate)
def resolve_with_args(self, text: str | None, group_id: str) -> Resolved | None:
"""Resolve a message in one group, with anything that followed it.
Two lookups, in this order. The whole text first, which is what every
command has always been matched by and stays byte-for-byte the same
rule -- so "!cat blah" still resolves to nothing. Only then the first
word, and only for a command that says it takes arguments.
A command the group is not allowed is not a near miss: it comes back
as None, indistinguishable from a word nobody registered.
"""
if not text:
return None
command = self.resolve(text)
if command is not None:
return Resolved(command, ()) if command.allowed_in(group_id) else None
words = text.split()
if not words:
return None
head = self._by_trigger.get(fold(words[0]))
if head is None or not head.takes_args or not head.allowed_in(group_id):
return None
return Resolved(head, tuple(words[1:]))
@property
def commands(self) -> list[Command]:
"""Every registered command, in registration order."""

16
signal_bot/transport_base.py
View File

@ -39,6 +39,22 @@ class Transport(Protocol):
a captioned photo is one message in the group rather than two.
"""
async def send_media(
self,
payload_base64: str,
recipient: str,
quote: Quote | None = None,
caption: str = "",
media_type: str = "",
filename: str = "",
) -> None:
"""Send an attachment of a stated type, optionally captioned.
send_image is the case where the type can be left to signal-cli.
Anything Signal renders differently -- a video, above all -- has to
say what it is, or it arrives as an unplayable blob.
"""
async def list_attachments(self) -> list[str]:
"""Return the ids of attachments stored server-side."""

21
signal_bot/transport_fake.py
View File

@ -26,6 +26,8 @@ class SentMessage:
image_base64: str | None = None
caption: str = ""
quote: Quote | None = None
media_type: str = ""
filename: str = ""
@dataclass
@ -58,9 +60,26 @@ class FakeTransport:
caption: str = "",
) -> None:
"""Record an image send."""
await self.send_media(image_base64, recipient, quote, caption)
async def send_media(
self,
payload_base64: str,
recipient: str,
quote: Quote | None = None,
caption: str = "",
media_type: str = "",
filename: str = "",
) -> None:
"""Record an attachment send, type and filename included."""
self.sent.append(
SentMessage(
recipient=recipient, image_base64=image_base64, quote=quote, caption=caption
recipient=recipient,
image_base64=payload_base64,
quote=quote,
caption=caption,
media_type=media_type,
filename=filename,
)
)

26
signal_bot/transport_rest.py
View File

@ -137,9 +137,33 @@ class RestTransport:
alongside base64_attachments comes back as one timestamp, so Signal
renders it as a single captioned photo rather than two messages.
"""
await self.send_media(image_base64, recipient, quote, caption)
async def send_media(
self,
payload_base64: str,
recipient: str,
quote: Quote | None = None,
caption: str = "",
media_type: str = "",
filename: str = "",
) -> None:
"""Send one attachment, naming its type when the caller knows it.
The data-URI spelling is the container's own, read off its
/swagger/doc.json rather than remembered: api.SendMessageV2 documents
base64_attachments as accepting bare base64,
"data:<MIME-TYPE>;base64,<DATA>" or
"data:<MIME-TYPE>;filename=<FILENAME>;base64,<DATA>". With no type
the bare form goes out, byte for byte what send_image always sent.
"""
attachment = payload_base64
if media_type:
named = f"filename={filename};" if filename else ""
attachment = f"data:{media_type};{named}base64,{payload_base64}"
payload: dict[str, Any] = {
"message": caption,
"base64_attachments": [image_base64],
"base64_attachments": [attachment],
"number": self._config.phone_number,
"recipients": [recipient],
}

2
signal_bot/versioning.py
View File

@ -30,7 +30,7 @@ PATCHES_PER_MINOR: Final = 8
# Which release this build is. Bump by one per release; everything else,
# including the version string in pyproject.toml, follows from it.
RELEASE_ORDINAL: Final = 16
RELEASE_ORDINAL: Final = 17
_POPES_FILE: Final = Path(__file__).with_name("popes.json")

15
tests/conftest.py
View File

@ -99,6 +99,21 @@ def _portraits_stay_out_of_the_real_cache(tmp_path: Path, monkeypatch: pytest.Mo
monkeypatch.setenv("PORTRAIT_CACHE_DIR", str(tmp_path / "portraits"))
@pytest.fixture(autouse=True)
def _no_test_ever_reads_a_real_photo_pack(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""Keep the whole suite away from the operator's private album.
The pack directory is discovered, not configured: with nothing set,
signal_bot.pack.default_dir falls back to $XDG_DATA_HOME and then to
$HOME, so a developer who has an album would have the suite load it --
quietly, and while still passing. Pointing XDG_DATA_HOME at an empty
temp dir makes "no pack" the answer everywhere unless a test builds one.
"""
monkeypatch.delenv("PHOTO_PACK_DIR", raising=False)
monkeypatch.delenv("STATE_DIRECTORY", raising=False)
monkeypatch.setenv("XDG_DATA_HOME", str(tmp_path / "share"))
@pytest.fixture
def store() -> Store:
with Store(":memory:") as opened:

78
tests/pack_fixture.py Normal file
View File

@ -0,0 +1,78 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""A synthetic photo pack, built in a temp directory.
Deliberately not a fixture: several test modules want packs of different
shapes, and a builder says which shape a test needs in the test itself. The
people are alfa, beta and gamma -- the real album's names are private and
must never appear in this repository, which is the whole reason the pack
lives outside it.
"""
from __future__ import annotations
import base64
import json
from typing import TYPE_CHECKING, Any
from tests.conftest import WATCHED_GROUP
if TYPE_CHECKING:
from pathlib import Path
# A 1x1 PNG, inline so no binary file enters the repo -- the same trick
# fake_signal.state uses for the fixture image it serves.
PIXEL = base64.b64decode(
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg=="
)
# Not a playable file, and it does not need to be: nothing in the bot decodes
# a video, it only has to be named and typed correctly on the way out.
CLIP = b"\x00\x00\x00\x18ftypmp42fake"
PEOPLE: list[dict[str, Any]] = [
{"key": "p1", "triggers": ["!alfa", "!alfka"]},
{"key": "p2", "triggers": ["!beta"]},
{"key": "p3", "triggers": ["!gamma", "!gammą"]},
]
COMBOS: list[dict[str, Any]] = [
{"key": "c1", "triggers": ["!oboje"], "all_of": ["p1", "p2"]},
{"key": "c2", "triggers": ["!ktokolwiek"], "any_of": ["p2", "p3"]},
]
TAGS: dict[str, list[str]] = {
"alfa.png": ["p1"],
"both.png": ["p1", "p2"],
"gamma.png": ["p3"],
"clip.mp4": ["p1"],
}
def build(
root: Path,
*,
people: list[dict[str, Any]] | None = None,
combos: list[dict[str, Any]] | None = None,
tags: dict[str, list[str]] | None = None,
groups: list[str] | None = None,
media: dict[str, bytes] | None = None,
write_tags: bool = True,
) -> Path:
"""Write a pack into ``root`` and return the directory."""
manifest: dict[str, Any] = {
"groups": [WATCHED_GROUP] if groups is None else groups,
"people": PEOPLE if people is None else people,
"combos": COMBOS if combos is None else combos,
}
tagged = TAGS if tags is None else tags
root.mkdir(parents=True, exist_ok=True)
(root / "people.json").write_text(json.dumps(manifest), encoding="utf-8")
if write_tags:
(root / "tags.json").write_text(json.dumps(tagged), encoding="utf-8")
album = root / "media"
album.mkdir(exist_ok=True)
# `is None`, not `or`: an empty mapping is a pack whose files are gone,
# which is a case the loader has to refuse and a test has to be able to ask for.
default = {name: CLIP if name.endswith(".mp4") else PIXEL for name in tagged}
files = default if media is None else media
for name, payload in files.items():
(album / name).write_bytes(payload)
return root

4
tests/stats_seed.py
View File

@ -18,9 +18,9 @@ if TYPE_CHECKING:
from signal_bot.store import Store
GROUP = "group-a=="
MITHRA = ("u-ala", "Ala")
ALA = ("u-ala", "Ala")
KUHY = ("u-kuhy", "kuhy")
IGA = ("u-iga", "Basia")
BASIA = ("u-basia", "Basia")
@dataclass(frozen=True, slots=True)

37
tests/test_bot.py
View File

@ -4,6 +4,7 @@
from __future__ import annotations
import datetime as dt
from dataclasses import replace
from typing import TYPE_CHECKING
import httpx
@ -12,6 +13,7 @@ from signal_bot.bot import Bot, build_bot, now_local
from signal_bot.liveness import Liveness
from signal_bot.stats.events import MessageEvent
from signal_bot.stats.periods import WARSAW
from tests import pack_fixture
from tests.conftest import (
WATCHED_GROUP,
envelope,
@ -19,14 +21,16 @@ from tests.conftest import (
)
if TYPE_CHECKING:
from pathlib import Path
from signal_bot.config import Config
from signal_bot.discovery import GroupDirectory
from signal_bot.store import Store
from signal_bot.transport_fake import FakeTransport
# The report covers the previous full day, so the tests freeze "now" and
# write their events into the day before it.
# The report covers the day it runs on, so the tests freeze "now" and write
# their events into it.
NOW = dt.datetime(2026, 9, 2, 21, 37, tzinfo=WARSAW)
YESTERDAY = "2026-09-01"
@ -78,3 +82,32 @@ async def test_listen_dispatches_every_frame(
"SELECT COUNT(*) AS n FROM message_events WHERE group_id = ?", (WATCHED_GROUP,)
).fetchone()
assert row["n"] == 1
def test_an_album_is_wired_up_only_where_there_is_one(
tmp_path: Path,
config: Config,
store: Store,
transport: FakeTransport,
directory: GroupDirectory,
) -> None:
"""Absent by default; present, and scoped, when the config names a pack."""
client = mock_client(lambda r: httpx.Response(200))
liveness = Liveness(config, client)
plain = build_bot(config, client, store, transport, directory, liveness)
assert plain.dispatcher.registry.resolve("!alfa") is None
root = pack_fixture.build(tmp_path / "pack")
with_album = build_bot(
replace(config, photo_pack_dir=root),
client,
store,
transport,
directory,
liveness,
)
command = with_album.dispatcher.registry.resolve("!alfa")
assert command is not None
assert command.name == "pack:p1"
assert command.groups == frozenset({WATCHED_GROUP})

81
tests/test_config_pack.py Normal file
View File

@ -0,0 +1,81 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Finding the private album, and the two different ways it can be absent.
Split out of test_config.py, which reached the 250-line cap.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
import pytest
from signal_bot.config import ConfigError
from signal_bot.config_env import load_config
from tests import pack_fixture
from tests.test_config import BASE_ENV
if TYPE_CHECKING:
from pathlib import Path
def test_an_installation_with_no_album_has_no_pack(tmp_path: Path) -> None:
"""Which is every machine but one, and every test in this suite."""
assert load_config(dict(BASE_ENV)).photo_pack_dir is None
def test_a_named_album_is_loaded_and_kept(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
root = pack_fixture.build(tmp_path / "pack")
monkeypatch.setenv("PHOTO_PACK_DIR", str(root))
assert load_config(dict(BASE_ENV)).photo_pack_dir == root
def test_a_named_album_that_is_not_there_fails_startup(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A typo in a path is an answer, not an outage worth retrying."""
monkeypatch.setenv("PHOTO_PACK_DIR", str(tmp_path / "nowhere"))
with pytest.raises(ConfigError, match="is not a directory"):
load_config(dict(BASE_ENV))
def test_a_broken_album_fails_startup_rather_than_every_command(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Parsed here, where load_config raising exits, not inside a retry loop."""
root = pack_fixture.build(tmp_path / "pack")
(root / "people.json").write_text("{}", encoding="utf-8")
monkeypatch.setenv("PHOTO_PACK_DIR", str(root))
with pytest.raises(ConfigError, match="defines no commands"):
load_config(dict(BASE_ENV))
def test_the_album_may_also_be_found_where_systemd_puts_it(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
pack_fixture.build(tmp_path / "state" / "pack")
monkeypatch.setenv("STATE_DIRECTORY", str(tmp_path / "state"))
assert load_config(dict(BASE_ENV)).photo_pack_dir == tmp_path / "state" / "pack"
def test_an_empty_album_directory_is_not_a_crash_loop(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""The tagger creates the directory before the manifest exists.
With Restart=always, failing on a directory somebody merely created would
put the unit into a restart loop -- so the implicit path is a hint, and
only a pack named outright is allowed to fail startup.
"""
(tmp_path / "share" / "signal-bot" / "pack" / "media").mkdir(parents=True)
monkeypatch.setenv("XDG_DATA_HOME", str(tmp_path / "share"))
assert load_config(dict(BASE_ENV)).photo_pack_dir is None
def test_a_named_album_with_no_manifest_fails_startup(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
(tmp_path / "pack").mkdir()
monkeypatch.setenv("PHOTO_PACK_DIR", str(tmp_path / "pack"))
with pytest.raises(ConfigError, match=r"has no people\.json"):
load_config(dict(BASE_ENV))

66
tests/test_dispatcher_gates.py
View File

@ -9,7 +9,8 @@ from typing import TYPE_CHECKING
from signal_bot.decorators import cooldown
from signal_bot.dispatcher import Dispatcher
from signal_bot.registry import CommandContext, CommandRegistry
from tests.conftest import WATCHED_GROUP, envelope
from signal_bot.stats.queries_day import day_rows
from tests.conftest import SECOND_GROUP, WATCHED_GROUP, envelope
if TYPE_CHECKING:
from signal_bot.config import Config
@ -85,3 +86,66 @@ async def test_waiting_the_stated_time_lets_the_command_run(
clock[0] += dt.timedelta(seconds=waited)
await dispatcher.handle(envelope(text="!cat"))
assert runs == ["cat", "cat"]
async def test_a_scoped_command_in_another_group_is_not_a_command_at_all(
config: Config, directory: GroupDirectory, store: Store, transport: FakeTransport
) -> None:
"""The whole privacy guarantee, at the only layer that can enforce it.
Not answered, not counted as a command, and -- because the dispatcher
never resolves it -- never named in a log line either. A handler that
simply returned would still leak all three.
"""
runs: list[str] = []
async def handler(ctx: CommandContext) -> None:
runs.append(ctx.command_name)
registry = CommandRegistry()
registry.register(
"pack:p1", ["!alfa"], handler, "album", groups=frozenset({SECOND_GROUP}), takes_args=True
)
dispatcher = Dispatcher(
config=config,
directory=directory,
registry=registry,
store=store,
transport=transport,
clock=lambda: NOW,
)
await dispatcher.handle(envelope(text="!alfa"))
assert runs == []
assert transport.sent == []
# Counted as an ordinary message, which is the other half of the claim:
# the text was seen, it simply was not a command here. The day comes from
# the message's own timestamp, never from the clock.
day = store.events.first_day(WATCHED_GROUP)
assert day is not None
assert [row.commands for row in day_rows(store.events, WATCHED_GROUP, day)] == [0]
async def test_the_words_after_a_trigger_reach_the_handler(
config: Config, directory: GroupDirectory, store: Store, transport: FakeTransport
) -> None:
seen: list[tuple[str, ...]] = []
async def handler(ctx: CommandContext) -> None:
seen.append(ctx.args)
registry = CommandRegistry()
registry.register("pack:p1", ["!alfa"], handler, "album", takes_args=True)
dispatcher = Dispatcher(
config=config,
directory=directory,
registry=registry,
store=store,
transport=transport,
clock=lambda: NOW,
)
await dispatcher.handle(envelope(text="!alfa beta gamma"))
assert seen == [("beta", "gamma")]

61
tests/test_help_command.py
View File

@ -14,19 +14,25 @@ from signal_bot.help_command import (
COMMAND_NAME,
GROUPS,
HEADING,
PACK_HEADING,
SPELLINGS,
register_help_command,
render_help,
)
from signal_bot.http_client import ImageFetcher
from signal_bot.models import parse_envelope
from signal_bot.pack_command import register_pack_commands
from signal_bot.pack_load import load as load_pack
from signal_bot.popes_command import register_pope_command
from signal_bot.portraits import PortraitCache
from signal_bot.registry import CommandContext, CommandRegistry
from signal_bot.stats.commands import register_stats_commands
from tests.conftest import WATCHED_GROUP, envelope
from tests import pack_fixture
from tests.conftest import SECOND_GROUP, WATCHED_GROUP, envelope
if TYPE_CHECKING:
from pathlib import Path
from signal_bot.config import Config
from signal_bot.store import Store
from signal_bot.transport_fake import FakeTransport
@ -73,14 +79,14 @@ def test_help_lists_every_command_exactly_once() -> None:
happens to mention "!stats" can neither satisfy nor break it.
"""
registry = full_registry()
text = render_help(registry.commands)
text = render_help(registry.commands, WATCHED_GROUP)
for command in registry.commands:
assert text.count(f"\n{command.primary} — {command.description}") == 1, command.name
def test_help_prints_the_polish_spelling_not_the_folded_one() -> None:
"""`triggers` holds skeletons; only `primary` still carries the diacritics."""
text = render_help(full_registry().commands)
text = render_help(full_registry().commands, WATCHED_GROUP)
assert "!papież" in text
assert "!papiez" not in text
assert "!tydzień" in text
@ -88,7 +94,7 @@ def test_help_prints_the_polish_spelling_not_the_folded_one() -> None:
def test_headings_appear_in_order() -> None:
"""Obrazki, Papież, Statystyki, then the ungrouped block, then the note."""
text = render_help(full_registry().commands)
text = render_help(full_registry().commands, WATCHED_GROUP)
positions = [text.index(f"\n{title}\n") for title, _ in GROUPS]
assert positions == sorted(positions)
assert text.startswith(HEADING)
@ -99,7 +105,7 @@ def test_a_command_outside_the_groups_is_still_listed() -> None:
"""The safety net: forgetting to add a command to GROUPS cannot hide it."""
registry = full_registry()
registry.register("ping", ["!ping"], _noop, "odpowiada pong")
assert "\n!ping — odpowiada pong" in render_help(registry.commands)
assert "\n!ping — odpowiada pong" in render_help(registry.commands, WATCHED_GROUP)
def test_a_group_naming_an_unknown_command_fails_loudly() -> None:
@ -107,7 +113,7 @@ def test_a_group_naming_an_unknown_command_fails_loudly() -> None:
registry = CommandRegistry()
register_help_command(registry)
with pytest.raises(KeyError):
render_help(registry.commands)
render_help(registry.commands, WATCHED_GROUP)
async def test_the_handler_replies_with_the_rendered_list(
@ -133,7 +139,7 @@ async def test_the_handler_replies_with_the_rendered_list(
)
)
assert transport.sent[0].recipient == WATCHED_GROUP
assert transport.sent[0].text == render_help(registry.commands)
assert transport.sent[0].text == render_help(registry.commands, WATCHED_GROUP)
@pytest.mark.parametrize("trigger", ["!pomoc", "!pomocy", "!help", "!komendy", "!commands"])
@ -142,3 +148,44 @@ def test_every_alias_resolves(trigger: str) -> None:
command = full_registry().resolve(trigger)
assert command is not None
assert command.name == COMMAND_NAME
def test_an_album_gets_its_own_heading_and_a_hint(tmp_path: Path) -> None:
"""And the hint is built from real triggers, because they are the example."""
registry = full_registry()
register_pack_commands(registry, load_pack(pack_fixture.build(tmp_path / "pack")))
text = render_help(registry.commands, WATCHED_GROUP)
assert f"\n{PACK_HEADING}\n" in text
assert "!alfa — losowe zdjęcie tej osoby" in text
assert "Kilka naraz (!alfa !beta)" in text
def test_the_album_is_invisible_in_every_other_group(tmp_path: Path) -> None:
"""The list is what would announce a private command family to strangers."""
registry = full_registry()
register_pack_commands(registry, load_pack(pack_fixture.build(tmp_path / "pack")))
text = render_help(registry.commands, SECOND_GROUP)
assert PACK_HEADING not in text
assert "alfa" not in text
assert "!kot" in text
def test_a_pack_of_one_prints_no_hint(tmp_path: Path) -> None:
""" "Combine them" needs two things to combine."""
root = pack_fixture.build(
tmp_path / "pack",
people=[{"key": "p1", "triggers": ["!alfa"]}],
combos=[],
tags={},
)
registry = full_registry()
register_pack_commands(registry, load_pack(root))
text = render_help(registry.commands, WATCHED_GROUP)
assert PACK_HEADING in text
assert "Kilka naraz" not in text

110
tests/test_pack.py Normal file
View File

@ -0,0 +1,110 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""The shape of a pack: where it lives, and what a typed name can mean."""
from __future__ import annotations
from pathlib import Path
from typing import TYPE_CHECKING
import pytest
from signal_bot.pack import MAX_NAMES, candidate_keysets, default_dir, media_type, options_for
from signal_bot.pack_load import load
from tests import pack_fixture
if TYPE_CHECKING:
from signal_bot.pack import Entry, Pack
def _entry(pack: Pack, key: str) -> Entry:
return next(entry for entry in pack.entries if entry.key == key)
@pytest.fixture
def pack(tmp_path: Path) -> Pack:
return load(pack_fixture.build(tmp_path / "pack"))
def test_a_media_type_comes_from_the_suffix_case_insensitively() -> None:
assert media_type("a.JPG") == "image/jpeg"
assert media_type("clip.mp4") == "video/mp4"
def test_matching_wants_every_key_not_any_of_them(pack: Pack) -> None:
assert pack.matching(frozenset({"p1"})) == ("alfa.png", "both.png", "clip.mp4")
assert pack.matching(frozenset({"p1", "p2"})) == ("both.png",)
assert pack.matching(frozenset({"p2", "p3"})) == ()
def test_media_path_is_inside_the_album(pack: Pack) -> None:
assert pack.media_path("alfa.png") == pack.directory / "media" / "alfa.png"
def test_an_explicit_directory_is_taken_at_its_word(monkeypatch: pytest.MonkeyPatch) -> None:
"""And is reported as explicit, which is what makes its absence an error."""
monkeypatch.setenv("PHOTO_PACK_DIR", "/srv/album")
assert default_dir() == (Path("/srv/album"), True)
def test_systemd_beats_anything_under_home(monkeypatch: pytest.MonkeyPatch) -> None:
"""ProtectHome=read-only is why: StateDirectory is exposed regardless."""
monkeypatch.setenv("STATE_DIRECTORY", "/var/lib/signal-bot")
assert default_dir() == (Path("/var/lib/signal-bot/pack"), False)
def test_the_xdg_fallback_needs_no_environment(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("XDG_DATA_HOME", raising=False)
monkeypatch.setattr(Path, "home", lambda: Path("/home/somebody"))
assert default_dir() == (Path("/home/somebody/.local/share/signal-bot/pack"), False)
def test_a_name_resolves_with_or_without_the_bang(pack: Pack) -> None:
""" "!alfa beta" is how people write it; the second half has no "!"."""
assert options_for(pack, "beta") == (frozenset({"p2"}),)
assert options_for(pack, "!beta") == (frozenset({"p2"}),)
def test_a_name_resolves_through_the_same_folding_as_a_command(pack: Pack) -> None:
"""The diacritic spelling is a separate alias and folds onto the same key."""
assert options_for(pack, "gammą") == (frozenset({"p3"}),)
def test_a_word_that_is_not_a_name_resolves_to_nothing(pack: Pack) -> None:
assert options_for(pack, "kto") is None
def test_one_trigger_alone_asks_for_that_person(pack: Pack) -> None:
assert candidate_keysets(pack, _entry(pack, "p1"), ()) == (frozenset({"p1"}),)
def test_a_tail_of_names_is_intersected(pack: Pack) -> None:
"""Which is the whole feature: "!alfa beta" means a photo with both."""
assert candidate_keysets(pack, _entry(pack, "p1"), ["beta"]) == (frozenset({"p1", "p2"}),)
def test_an_all_of_combo_is_one_set_of_several_people(pack: Pack) -> None:
assert candidate_keysets(pack, _entry(pack, "c1"), ()) == (frozenset({"p1", "p2"}),)
def test_an_any_of_combo_offers_each_reading_separately(pack: Pack) -> None:
"""So the handler can prefer one that actually has a photo."""
assert candidate_keysets(pack, _entry(pack, "c2"), ()) == (
frozenset({"p2"}),
frozenset({"p3"}),
)
def test_a_choice_multiplies_out_with_the_rest_of_the_line(pack: Pack) -> None:
assert candidate_keysets(pack, _entry(pack, "c2"), ["alfa"]) == (
frozenset({"p2", "p1"}),
frozenset({"p3", "p1"}),
)
def test_an_unknown_word_in_the_tail_is_not_silently_dropped(pack: Pack) -> None:
"""Dropping it would answer a different question and look like bad tags."""
assert candidate_keysets(pack, _entry(pack, "p1"), ["beta", "nikt"]) is None
def test_too_many_names_is_refused_rather_than_multiplied_out(pack: Pack) -> None:
assert candidate_keysets(pack, _entry(pack, "p1"), ["beta"] * MAX_NAMES) is None

183
tests/test_pack_command.py Normal file
View File

@ -0,0 +1,183 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Answering one photo request: what gets sent, and what gets refused."""
from __future__ import annotations
import base64
from typing import TYPE_CHECKING
import pytest
from signal_bot.models import parse_envelope
from signal_bot.pack_command import (
EITHER,
EVERYBODY,
NO_PHOTO,
ONE_PERSON,
description_for,
register_pack_commands,
)
from signal_bot.pack_load import load
from signal_bot.registry import CommandContext, CommandRegistry
from tests import pack_fixture
from tests.conftest import WATCHED_GROUP, envelope
if TYPE_CHECKING:
from pathlib import Path
from signal_bot.config import Config
from signal_bot.pack import Pack
from signal_bot.registry import Handler
from signal_bot.store import Store
from signal_bot.transport_fake import FakeTransport
@pytest.fixture
def pack(tmp_path: Path) -> Pack:
return load(pack_fixture.build(tmp_path / "pack"))
def _registry(pack: Pack) -> CommandRegistry:
registry = CommandRegistry()
register_pack_commands(registry, pack)
return registry
def _ctx(
transport: FakeTransport,
store: Store,
config: Config,
text: str,
args: tuple[str, ...] = (),
) -> CommandContext:
message = parse_envelope(envelope(text=text))
assert message is not None
return CommandContext(
message=message,
group_id=WATCHED_GROUP,
transport=transport,
store=store,
config=config,
command_name="pack:p1",
now=0.0,
day="2026-08-30",
args=args,
)
async def _run(
pack: Pack, transport: FakeTransport, store: Store, config: Config, text: str
) -> None:
"""Type one line at the pack, the way the dispatcher would."""
registry = _registry(pack)
resolved = registry.resolve_with_args(text, WATCHED_GROUP)
assert resolved is not None
await resolved.command.handler(_ctx(transport, store, config, text, resolved.args))
def test_each_entry_shape_explains_itself(pack: Pack) -> None:
by_key = {entry.key: entry for entry in pack.entries}
assert description_for(by_key["p1"]) == ONE_PERSON
assert description_for(by_key["c1"]) == EVERYBODY
assert description_for(by_key["c2"]) == EITHER
def test_a_description_never_names_anybody(pack: Pack) -> None:
"""!pomoc prints it; the trigger is the only place a name may appear."""
for entry in pack.entries:
described = description_for(entry)
assert described in {ONE_PERSON, EVERYBODY, EITHER}
assert not any(trigger.strip("!") in described for trigger in entry.triggers)
async def test_a_photo_is_sent_with_its_own_type_and_name(
pack: Pack, transport: FakeTransport, store: Store, config: Config
) -> None:
await _run(pack, transport, store, config, "!gamma")
sent = transport.sent[0]
assert sent.filename == "gamma.png"
assert sent.media_type == "image/png"
assert sent.image_base64 is not None
assert base64.b64decode(sent.image_base64) == pack_fixture.PIXEL
assert sent.quote is not None
async def test_two_names_want_a_photo_with_both(
pack: Pack, transport: FakeTransport, store: Store, config: Config
) -> None:
"""both.png is the only file tagged with p1 and p2 at once."""
await _run(pack, transport, store, config, "!alfa beta")
assert transport.sent[0].filename == "both.png"
async def test_the_second_name_may_carry_its_own_bang(
pack: Pack, transport: FakeTransport, store: Store, config: Config
) -> None:
await _run(pack, transport, store, config, "!alfa !beta")
assert transport.sent[0].filename == "both.png"
async def test_a_combination_nobody_photographed_is_answered_in_one_line(
pack: Pack, transport: FakeTransport, store: Store, config: Config
) -> None:
await _run(pack, transport, store, config, "!beta gamma")
assert transport.texts == [NO_PHOTO]
assert transport.sent[0].image_base64 is None
async def test_a_name_nobody_has_gets_the_same_answer(
pack: Pack, transport: FakeTransport, store: Store, config: Config
) -> None:
"""From the group's side it is the same question: is there such a photo?"""
await _run(pack, transport, store, config, "!alfa nikt")
assert transport.texts == [NO_PHOTO]
async def test_a_choice_lands_on_the_reading_that_has_a_photo(
tmp_path: Path, transport: FakeTransport, store: Store, config: Config
) -> None:
"""p2 has no photo of their own, so "either of them" must not fail."""
pack = load(
pack_fixture.build(tmp_path / "pack", tags={"gamma.png": ["p3"], "both.png": ["p1", "p2"]})
)
for _ in range(5):
await _run(pack, transport, store, config, "!ktokolwiek")
assert {message.filename for message in transport.sent} == {"gamma.png", "both.png"}
async def test_a_video_keeps_being_a_video(
tmp_path: Path, transport: FakeTransport, store: Store, config: Config
) -> None:
"""Told it is a jpeg, Signal renders an unplayable file."""
pack = load(pack_fixture.build(tmp_path / "pack", tags={"clip.mp4": ["p1"]}))
await _run(pack, transport, store, config, "!alfa")
assert transport.sent[0].media_type == "video/mp4"
assert transport.sent[0].filename == "clip.mp4"
def test_every_command_is_named_by_key_and_scoped_to_the_pack(pack: Pack) -> None:
"""A log line or a command_uses row must not say who was asked for."""
registry = _registry(pack)
assert [command.name for command in registry.commands] == [
"pack:p1",
"pack:p2",
"pack:p3",
"pack:c1",
"pack:c2",
]
for command in registry.commands:
assert command.groups == pack.groups
assert command.takes_args
def test_the_gate_is_applied_to_every_entry(pack: Pack) -> None:
registry = CommandRegistry()
marked: list[str] = []
def decorate(handler: Handler) -> Handler:
"""Count the wrap and change nothing else."""
marked.append("wrapped")
return handler
register_pack_commands(registry, pack, decorate=decorate)
assert len(marked) == len(pack.entries)

191
tests/test_pack_load.py Normal file
View File

@ -0,0 +1,191 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Everything a photo pack can get wrong, and the sentence it gets back.
Each of these fails startup rather than a command. A manifest that cannot be
honoured is an answer, not an outage: the alternative is a bot that comes up
healthy and answers "no such photo" forever because a file was renamed.
"""
from __future__ import annotations
import json
from typing import TYPE_CHECKING
import pytest
from signal_bot.groups import ConfigError
from signal_bot.pack_load import load
from tests import pack_fixture
if TYPE_CHECKING:
from pathlib import Path
from signal_bot.pack import Pack
def _load(root: Path, **kwargs: object) -> Pack:
"""Build a pack of the requested shape and load it."""
return load(pack_fixture.build(root / "pack", **kwargs))
def test_a_pack_loads_its_people_its_combos_and_its_tags(tmp_path: Path) -> None:
pack = _load(tmp_path)
assert [entry.key for entry in pack.entries] == ["p1", "p2", "p3", "c1", "c2"]
assert pack.tags["both.png"] == frozenset({"p1", "p2"})
assert pack.groups
def test_a_missing_manifest_says_which_file(tmp_path: Path) -> None:
with pytest.raises(ConfigError, match="cannot be read"):
load(tmp_path / "nowhere")
def test_a_manifest_that_is_not_json_says_so(tmp_path: Path) -> None:
root = pack_fixture.build(tmp_path / "pack")
(root / "people.json").write_text("{oops", encoding="utf-8")
with pytest.raises(ConfigError, match="not valid JSON"):
load(root)
def test_a_manifest_that_is_not_an_object_is_refused(tmp_path: Path) -> None:
root = pack_fixture.build(tmp_path / "pack")
(root / "people.json").write_text("[]", encoding="utf-8")
with pytest.raises(ConfigError, match="must be an object"):
load(root)
def test_a_pack_with_no_groups_is_refused(tmp_path: Path) -> None:
"""Defaulting to "every group" would put private photos in a stranger's chat."""
with pytest.raises(ConfigError, match="non-empty 'groups'"):
_load(tmp_path, groups=[])
def test_a_blank_group_id_is_refused(tmp_path: Path) -> None:
with pytest.raises(ConfigError, match="blank group id"):
_load(tmp_path, groups=[" "])
def test_people_have_to_be_objects(tmp_path: Path) -> None:
with pytest.raises(ConfigError, match="must be a list of objects"):
_load(tmp_path, people=["alfa"])
def test_a_pack_with_no_entries_at_all_is_refused(tmp_path: Path) -> None:
with pytest.raises(ConfigError, match="defines no commands"):
_load(tmp_path, people=[], combos=[], tags={})
def test_a_key_that_reads_like_a_name_is_refused(tmp_path: Path) -> None:
"""The key is what gets logged; it exists precisely so a name is not."""
with pytest.raises(ConfigError, match="alphanumeric 'key'"):
_load(tmp_path, people=[{"key": "", "triggers": ["!alfa"]}], combos=[], tags={})
def test_a_non_ascii_key_is_refused(tmp_path: Path) -> None:
with pytest.raises(ConfigError, match="alphanumeric 'key'"):
_load(tmp_path, people=[{"key": "ala", "triggers": ["!alfa"]}], combos=[], tags={})
def test_an_entry_needs_at_least_one_trigger(tmp_path: Path) -> None:
with pytest.raises(ConfigError, match="non-empty 'triggers'"):
_load(tmp_path, people=[{"key": "p1", "triggers": []}], combos=[], tags={})
def test_a_trigger_without_a_bang_could_never_fire(tmp_path: Path) -> None:
with pytest.raises(ConfigError, match='must start with "!"'):
_load(tmp_path, people=[{"key": "p1", "triggers": ["alfa"]}], combos=[], tags={})
def test_a_trigger_with_a_digit_is_refused(tmp_path: Path) -> None:
"""folding maps 0134578 onto letters, so "!alfa7" is not what you typed."""
with pytest.raises(ConfigError, match="must not contain a digit"):
_load(tmp_path, people=[{"key": "p1", "triggers": ["!alfa7"]}], combos=[], tags={})
def test_a_trigger_that_folds_away_to_nothing_is_refused(tmp_path: Path) -> None:
with pytest.raises(ConfigError, match="folds away to nothing"):
_load(tmp_path, people=[{"key": "p1", "triggers": ["!‍"]}], combos=[], tags={})
def test_a_combo_needs_exactly_one_of_all_of_and_any_of(tmp_path: Path) -> None:
with pytest.raises(ConfigError, match="exactly one of"):
_load(
tmp_path,
combos=[{"key": "c1", "triggers": ["!oboje"], "all_of": ["p1"], "any_of": ["p2"]}],
)
def test_a_combo_with_neither_is_refused(tmp_path: Path) -> None:
with pytest.raises(ConfigError, match="exactly one of"):
_load(tmp_path, combos=[{"key": "c1", "triggers": ["!oboje"]}])
def test_a_combo_naming_somebody_who_does_not_exist_is_refused(tmp_path: Path) -> None:
with pytest.raises(ConfigError, match="unknown people"):
_load(tmp_path, combos=[{"key": "c1", "triggers": ["!oboje"], "all_of": ["p9"]}])
def test_a_key_used_twice_is_refused(tmp_path: Path) -> None:
"""Two entries on one key would share a cooldown row and a log line."""
people = [{"key": "p1", "triggers": ["!alfa"]}, {"key": "p1", "triggers": ["!beta"]}]
with pytest.raises(ConfigError, match="twice"):
_load(tmp_path, people=people, combos=[], tags={})
def test_two_entries_folding_onto_one_trigger_are_refused(tmp_path: Path) -> None:
"""The registry would catch it too, but could not say which entries."""
people = [{"key": "p1", "triggers": ["!gamma"]}, {"key": "p2", "triggers": ["!gammą"]}]
with pytest.raises(ConfigError, match="claimed by"):
_load(tmp_path, people=people, combos=[], tags={})
def test_one_entry_may_spell_itself_two_ways_that_fold_the_same(tmp_path: Path) -> None:
"""Which is the point: "!ala" and "!ala" are one person, written twice."""
pack = _load(
tmp_path,
people=[{"key": "p1", "triggers": ["!gamma", "!gammą"]}],
combos=[],
tags={},
)
assert len(pack.entries) == 1
def test_an_untagged_pack_loads_and_warns(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
"""Not an error: that is an album waiting for the tagger, not a broken one."""
pack = _load(tmp_path, write_tags=False)
assert pack.tags == {}
assert "tagger" in caplog.text
def test_a_tag_naming_a_file_that_is_gone_is_refused(tmp_path: Path) -> None:
"""A renamed file would otherwise drop out of the album in silence."""
with pytest.raises(ConfigError, match="not a file"):
_load(tmp_path, tags={"alfa.png": ["p1"]}, media={})
def test_a_tag_may_not_reach_outside_the_album(tmp_path: Path) -> None:
with pytest.raises(ConfigError, match="not a file"):
_load(tmp_path, tags={"../people.json": ["p1"]}, media={"alfa.png": pack_fixture.PIXEL})
def test_a_tag_on_a_type_signal_is_not_told_about_is_refused(tmp_path: Path) -> None:
with pytest.raises(ConfigError, match="not one Signal is told about"):
_load(tmp_path, tags={"notes.txt": ["p1"]}, media={"notes.txt": b"hello"})
def test_a_tag_naming_somebody_outside_the_manifest_is_refused(tmp_path: Path) -> None:
with pytest.raises(ConfigError, match=r"names somebody who is not in"):
_load(tmp_path, tags={"alfa.png": ["p9"]})
def test_a_tag_that_is_not_a_list_is_refused(tmp_path: Path) -> None:
root = pack_fixture.build(tmp_path / "pack")
(root / "tags.json").write_text(json.dumps({"alfa.png": "p1"}), encoding="utf-8")
with pytest.raises(ConfigError, match="names somebody who is not in"):
load(root)
def test_an_untagged_file_is_simply_not_in_the_album(tmp_path: Path) -> None:
"""The tagger writes an empty list for "seen, nobody in it"."""
pack = _load(tmp_path, tags={"alfa.png": [], "both.png": ["p1", "p2"]})
assert set(pack.tags) == {"both.png"}

46
tests/test_registry.py
View File

@ -89,3 +89,49 @@ async def test_reply_image_passes_the_caption_through(store: Store, config: Conf
assert sent.caption == "8. Telesfor"
assert sent.image_base64 == "Zm9v"
assert sent.quote is not None
async def _nothing(ctx: CommandContext) -> None:
"""A handler that does nothing, for tests about resolution only."""
def test_a_command_with_no_groups_exists_everywhere() -> None:
registry = CommandRegistry()
command = registry.register("open", ["!open"], _nothing, "everywhere")
assert command.allowed_in("group-a==")
assert registry.resolve_with_args("!open", "group-b==") is not None
def test_a_scoped_command_does_not_exist_elsewhere() -> None:
"""Not refused -- absent. A refusal would announce that it exists."""
registry = CommandRegistry()
registry.register("shy", ["!shy"], _nothing, "here only", groups=frozenset({"group-a=="}))
assert registry.resolve_with_args("!shy", "group-a==") is not None
assert registry.resolve_with_args("!shy", "group-b==") is None
def test_words_after_a_trigger_are_arguments_only_where_they_are_wanted() -> None:
""" "!cat blah" has always resolved to nothing, and still does."""
registry = CommandRegistry()
registry.register("plain", ["!plain"], _nothing, "no arguments")
registry.register("takes", ["!takes"], _nothing, "arguments", takes_args=True)
assert registry.resolve_with_args("!plain blah", "g==") is None
resolved = registry.resolve_with_args("!takes one two", "g==")
assert resolved is not None
assert resolved.args == ("one", "two")
def test_a_bare_trigger_has_no_arguments() -> None:
registry = CommandRegistry()
registry.register("takes", ["!takes"], _nothing, "arguments", takes_args=True)
resolved = registry.resolve_with_args("!takes", "g==")
assert resolved is not None
assert resolved.args == ()
def test_nothing_resolves_out_of_an_empty_message() -> None:
registry = CommandRegistry()
registry.register("takes", ["!takes"], _nothing, "arguments", takes_args=True)
assert registry.resolve_with_args(None, "g==") is None
assert registry.resolve_with_args(" ", "g==") is None
assert registry.resolve_with_args("!nobody has this", "g==") is None

28
tests/test_stats_commands.py
View File

@ -14,7 +14,7 @@ from signal_bot.registry import CommandRegistry
from signal_bot.stats import periods
from signal_bot.stats.backfill import BACKFILL_KEY, seed_from_counter
from signal_bot.stats.commands import HANDLERS, register_stats_commands
from tests.stats_seed import GROUP, IGA, KUHY, MITHRA, Shape, wrote
from tests.stats_seed import ALA, BASIA, GROUP, KUHY, Shape, wrote
from tests.test_stats_backfill import LEGACY as LEGACY_COUNTER
if TYPE_CHECKING:
@ -100,7 +100,7 @@ async def test_every_command_answers_on_an_empty_history(store: Store, name: str
async def test_today_and_yesterday_are_different_days(store: Store) -> None:
wrote(store, TODAY, MITHRA, 4)
wrote(store, TODAY, ALA, 4)
wrote(store, YESTERDAY, KUHY, 7)
assert "Dziś, na razie:" in await run(store, "stats_today")
assert "Ala" in await run(store, "stats_today")
@ -109,7 +109,7 @@ async def test_today_and_yesterday_are_different_days(store: Store) -> None:
async def test_yesterday_renders_the_nightly_block_verbatim(store: Store) -> None:
"""The only way to check the headline output without waiting for 21:37."""
wrote(store, YESTERDAY, MITHRA, 5)
wrote(store, YESTERDAY, ALA, 5)
answer = await run(store, "stats_yesterday")
assert answer.startswith("📊 ")
assert "👑 Ala — 5" in answer
@ -120,7 +120,7 @@ async def test_a_day_nobody_wrote_on_says_so(store: Store) -> None:
async def test_the_week_shows_a_share_bar(store: Store) -> None:
wrote(store, TODAY, MITHRA, 8)
wrote(store, TODAY, ALA, 8)
wrote(store, TODAY, KUHY, 2)
answer = await run(store, "stats_week")
assert "Ala — 8" in answer
@ -129,9 +129,9 @@ async def test_the_week_shows_a_share_bar(store: Store) -> None:
async def test_the_month_puts_medals_on_the_podium(store: Store) -> None:
wrote(store, TODAY, MITHRA, 8)
wrote(store, TODAY, ALA, 8)
wrote(store, TODAY, KUHY, 5)
wrote(store, TODAY, IGA, 2)
wrote(store, TODAY, BASIA, 2)
answer = await run(store, "stats_month")
assert "🥇 Ala" in answer
assert "🥈 kuhy" in answer
@ -139,13 +139,13 @@ async def test_the_month_puts_medals_on_the_podium(store: Store) -> None:
async def test_a_first_period_says_so_rather_than_claiming_no_change(store: Store) -> None:
wrote(store, TODAY, MITHRA, 8)
wrote(store, TODAY, ALA, 8)
assert "pierwszy taki okres w historii" in await run(store, "stats_year")
async def test_me_reports_the_asker(store: Store) -> None:
wrote(store, TODAY, KUHY, 6, Shape(hour=23))
wrote(store, TODAY, MITHRA, 2)
wrote(store, TODAY, ALA, 2)
answer = await run(store, "stats_me")
assert "kuhy — 6 wiadomości w historii" in answer
assert "Udział w rozmowie: 75%" in answer
@ -153,13 +153,13 @@ async def test_me_reports_the_asker(store: Store) -> None:
async def test_me_is_honest_with_somebody_who_never_wrote(store: Store) -> None:
wrote(store, TODAY, MITHRA, 4)
wrote(store, TODAY, ALA, 4)
assert "ani jednej wiadomości" in await run(store, "stats_me", uuid="u-nobody")
async def test_records_and_crown_fill_in_once_there_is_history(store: Store) -> None:
wrote(store, periods.shift(TODAY, -2), MITHRA, 9)
wrote(store, periods.shift(TODAY, -1), MITHRA, 4)
wrote(store, periods.shift(TODAY, -2), ALA, 9)
wrote(store, periods.shift(TODAY, -1), ALA, 4)
wrote(store, TODAY, KUHY, 3)
records = await run(store, "stats_records")
assert "Najwięcej w jeden dzień: 9" in records
@ -177,8 +177,8 @@ async def test_records_skips_somebody_with_no_days_left_after_exclusions(
Fixed dates, not periods.today(): on the day this was written those were
the same date, and the test passed for the wrong reason.
"""
wrote(store, "2026-08-01", IGA, 40)
wrote(store, "2026-08-15", MITHRA, 5)
wrote(store, "2026-08-01", BASIA, 40)
wrote(store, "2026-08-15", ALA, 5)
store.set_meta(GROUP, BACKFILL_KEY, "2026-08-01")
rendered = await run(store, "stats_records")
assert "Ala" in rendered
@ -211,5 +211,5 @@ async def test_somebody_with_only_backfilled_messages_has_no_hour(
async def test_weekdays_marks_today(store: Store) -> None:
wrote(store, TODAY, MITHRA, 5)
wrote(store, TODAY, ALA, 5)
assert "← dziś" in await run(store, "stats_weekdays")

32
tests/test_stats_curiosities.py
View File

@ -9,7 +9,7 @@ from signal_bot.stats import curiosities
from signal_bot.stats.facts import DayFacts, gather
from signal_bot.stats.highlights import Highlight, ratio, select
from signal_bot.stats.queries_day import day_rows, with_silent
from tests.stats_seed import GROUP, IGA, KUHY, MITHRA, Shape, wrote
from tests.stats_seed import ALA, BASIA, GROUP, KUHY, Shape, wrote
if TYPE_CHECKING:
from signal_bot.store import Store
@ -29,17 +29,17 @@ def test_day_one_notices_nothing(store: Store) -> None:
def test_facts_expose_the_days_crown_holders(store: Store) -> None:
wrote(store, "2026-09-01", MITHRA, 4)
wrote(store, "2026-09-01", ALA, 4)
wrote(store, "2026-09-01", KUHY, 4)
wrote(store, "2026-09-01", IGA, 1)
wrote(store, "2026-09-01", BASIA, 1)
assert sorted(n for _, n in facts_for(store, "2026-09-01").leaders) == ["Ala", "kuhy"]
assert facts_for(store, "2026-09-09").leaders == []
def test_a_personal_record_reports_what_it_beat(store: Store) -> None:
wrote(store, "2026-09-01", MITHRA, 5)
wrote(store, "2026-09-01", ALA, 5)
wrote(store, "2026-09-02", KUHY, 1)
wrote(store, "2026-09-02", MITHRA, 9)
wrote(store, "2026-09-02", ALA, 9)
found = curiosities.personal_record(facts_for(store, "2026-09-02"))
assert found is not None
assert "Ala 9 — rekord życiowy (poprzedni 5, 1 września)" in found.text
@ -47,7 +47,7 @@ def test_a_personal_record_reports_what_it_beat(store: Store) -> None:
def test_a_first_ever_day_is_not_announced_as_a_record(store: Store) -> None:
"""Everybody's first day is their best; saying so every time is noise."""
wrote(store, "2026-09-01", MITHRA, 5)
wrote(store, "2026-09-01", ALA, 5)
assert curiosities.personal_record(facts_for(store, "2026-09-01")) is None
@ -75,10 +75,10 @@ def test_a_media_record_needs_more_than_one_photo(store: Store) -> None:
def test_a_crown_streak_needs_two_days(store: Store) -> None:
wrote(store, "2026-09-01", MITHRA, 5)
wrote(store, "2026-09-01", ALA, 5)
wrote(store, "2026-09-01", KUHY, 1)
assert curiosities.crown_streak(facts_for(store, "2026-09-01")) is None
wrote(store, "2026-09-02", MITHRA, 5)
wrote(store, "2026-09-02", ALA, 5)
wrote(store, "2026-09-02", KUHY, 1)
found = curiosities.crown_streak(facts_for(store, "2026-09-02"))
assert found is not None
@ -87,7 +87,7 @@ def test_a_crown_streak_needs_two_days(store: Store) -> None:
def test_a_shared_crown_names_everyone(store: Store) -> None:
for day in ("2026-09-01", "2026-09-02"):
wrote(store, day, MITHRA, 4)
wrote(store, day, ALA, 4)
wrote(store, day, KUHY, 4)
found = curiosities.crown_streak(facts_for(store, "2026-09-02"))
assert found is not None
@ -95,10 +95,10 @@ def test_a_shared_crown_names_everyone(store: Store) -> None:
def test_breaking_a_silence(store: Store) -> None:
wrote(store, "2026-09-01", IGA, 2)
wrote(store, "2026-09-01", BASIA, 2)
for day in ("2026-09-02", "2026-09-03", "2026-09-04"):
wrote(store, day, KUHY, 3)
wrote(store, "2026-09-05", IGA, 1)
wrote(store, "2026-09-05", BASIA, 1)
wrote(store, "2026-09-05", KUHY, 9)
found = curiosities.silence_broken(facts_for(store, "2026-09-05"))
assert found is not None
@ -106,9 +106,9 @@ def test_breaking_a_silence(store: Store) -> None:
def test_a_short_gap_is_not_worth_mentioning(store: Store) -> None:
wrote(store, "2026-09-01", IGA, 2)
wrote(store, "2026-09-01", BASIA, 2)
wrote(store, "2026-09-02", KUHY, 3)
wrote(store, "2026-09-03", IGA, 2)
wrote(store, "2026-09-03", BASIA, 2)
assert curiosities.silence_broken(facts_for(store, "2026-09-03")) is None
@ -125,7 +125,7 @@ def test_one_single_message_is_not_a_monologue(store: Store) -> None:
def test_dominating_the_day(store: Store) -> None:
wrote(store, "2026-09-01", MITHRA, 8)
wrote(store, "2026-09-01", ALA, 8)
wrote(store, "2026-09-01", KUHY, 2)
found = curiosities.dominant_share(facts_for(store, "2026-09-01"))
assert found is not None
@ -133,9 +133,9 @@ def test_dominating_the_day(store: Store) -> None:
def test_an_evenly_shared_day_is_not_dominated(store: Store) -> None:
wrote(store, "2026-09-01", MITHRA, 5)
wrote(store, "2026-09-01", ALA, 5)
wrote(store, "2026-09-01", KUHY, 4)
wrote(store, "2026-09-01", IGA, 4)
wrote(store, "2026-09-01", BASIA, 4)
assert curiosities.dominant_share(facts_for(store, "2026-09-01")) is None

8
tests/test_stats_queries_period.py
View File

@ -15,7 +15,7 @@ from signal_bot.stats.queries_period import (
total_for,
weekday_buckets,
)
from tests.stats_seed import GROUP, IGA, KUHY, MITHRA, Shape, wrote
from tests.stats_seed import ALA, BASIA, GROUP, KUHY, Shape, wrote
if TYPE_CHECKING:
from signal_bot.store import Store
@ -28,10 +28,10 @@ PREVIOUS = periods.previous(WEEK)
def history(store: Store) -> None:
"""A full week, plus one day in the week before it."""
wrote(store, "2026-08-25", KUHY, 8)
wrote(store, "2026-08-31", MITHRA, 10, Shape(media=2, commands=1))
wrote(store, "2026-09-01", MITHRA, 20)
wrote(store, "2026-08-31", ALA, 10, Shape(media=2, commands=1))
wrote(store, "2026-09-01", ALA, 20)
wrote(store, "2026-09-01", KUHY, 10)
wrote(store, "2026-09-02", IGA, 5)
wrote(store, "2026-09-02", BASIA, 5)
def test_period_rows_are_ordered_by_volume(store: Store) -> None:

18
tests/test_stats_records.py
View File

@ -15,7 +15,7 @@ from signal_bot.stats.records import (
previous_best,
quietest_day,
)
from tests.stats_seed import GROUP, IGA, KUHY, MITHRA, Shape, wrote
from tests.stats_seed import ALA, BASIA, GROUP, KUHY, Shape, wrote
if TYPE_CHECKING:
from signal_bot.store import Store
@ -23,11 +23,11 @@ if TYPE_CHECKING:
def history(store: Store) -> None:
"""Four days, a two-day gap, ties, and somebody who stops writing."""
wrote(store, "2026-09-01", MITHRA, 10, Shape(media=3))
wrote(store, "2026-09-01", ALA, 10, Shape(media=3))
wrote(store, "2026-09-01", KUHY, 5)
wrote(store, "2026-09-02", MITHRA, 4)
wrote(store, "2026-09-02", ALA, 4)
wrote(store, "2026-09-02", KUHY, 4)
wrote(store, "2026-09-03", IGA, 2)
wrote(store, "2026-09-03", BASIA, 2)
# 09-04 and 09-05 silent.
wrote(store, "2026-09-06", KUHY, 9, Shape(media=1))
@ -61,7 +61,7 @@ def test_the_quietest_day_is_one_somebody_spoke_on(store: Store) -> None:
def test_a_tie_on_the_busiest_day_goes_to_the_earlier(store: Store) -> None:
wrote(store, "2026-09-02", KUHY, 7)
wrote(store, "2026-09-01", MITHRA, 7)
wrote(store, "2026-09-01", ALA, 7)
assert busiest_day(every_day(store.events, GROUP)).day == "2026-09-01"
@ -75,8 +75,8 @@ def test_personal_best_and_what_it_beat(store: Store) -> None:
def test_a_first_ever_day_has_nothing_to_beat(store: Store) -> None:
"""The "poprzedni rekord" clause has to be omitted, not printed as 0."""
wrote(store, "2026-09-01", IGA, 3)
assert previous_best(store.events, GROUP, "u-iga", "2026-09-01") is None
wrote(store, "2026-09-01", BASIA, 3)
assert previous_best(store.events, GROUP, "u-basia", "2026-09-01") is None
def test_most_media_day(store: Store) -> None:
@ -129,8 +129,8 @@ def test_somebody_who_never_wrote_is_silent_for_every_day_before_now(store: Stor
def test_the_backfilled_day_is_kept_out_of_records(store: Store) -> None:
"""A 22-hour window filed as a day must not become somebody's best."""
wrote(store, "2026-08-30", MITHRA, 98)
wrote(store, "2026-09-01", MITHRA, 10)
wrote(store, "2026-08-30", ALA, 98)
wrote(store, "2026-09-01", ALA, 10)
assert personal_best(store.events, GROUP, "u-ala").value == 98
excluded = personal_best(store.events, GROUP, "u-ala", exclude="2026-08-30")
assert (excluded.day, excluded.value) == ("2026-09-01", 10)

8
tests/test_stats_render_day.py
View File

@ -34,7 +34,7 @@ def rows(spec: list[tuple[str, int, int]], *, commands: int = 0) -> list[DayRow]
def test_the_nightly_block_for_a_real_day() -> None:
"""Pinned exactly: this is the only assertion that proves the rendering."""
day_rows_ = with_silent(rows(GROUP), [("u9", "Marek")])
day_rows_ = with_silent(rows(GROUP), [("u9", "Celina")])
rendered = render_day(
"2026-09-01",
day_rows_,
@ -50,7 +50,7 @@ def test_the_nightly_block_for_a_real_day() -> None:
"5. Franciszek Kowalczyk — 16\n"
"6. Cezary Wiśniewski — 15\n"
"🏆 Ala 76 — rekord życiowy (poprzedni 71)\n"
"Cisza: Marek\n"
"Cisza: Celina\n"
"Średnio 47 znaków na wiadomość.\n"
"(liczymy od 30 sierpnia)"
)
@ -84,9 +84,9 @@ def test_summary_is_empty_on_a_day_nobody_spoke() -> None:
def test_unnamed_members_are_counted_never_named() -> None:
"""Their only spellings are a uuid or a phone number; both would leak."""
day = with_silent(rows([("kuhy", 3, 0)]), [("u9", "Marek")])
day = with_silent(rows([("kuhy", 3, 0)]), [("u9", "Celina")])
rendered = render_day("2026-09-01", day, day_totals(day), DayView(unnamed=2))
assert "Cisza: Marek, i 2 osoby bez ani jednej wiadomości tutaj" in rendered
assert "Cisza: Celina, i 2 osoby bez ani jednej wiadomości tutaj" in rendered
def test_a_big_group_is_truncated_and_still_fits() -> None:

42
tests/test_stats_streaks.py
View File

@ -12,7 +12,7 @@ from signal_bot.stats.streaks import (
longest_ever,
names_of,
)
from tests.stats_seed import GROUP, IGA, KUHY, MITHRA, wrote
from tests.stats_seed import ALA, BASIA, GROUP, KUHY, wrote
if TYPE_CHECKING:
from signal_bot.store import Store
@ -27,7 +27,7 @@ def test_no_history_has_no_crown(store: Store) -> None:
def test_the_busiest_person_holds_the_day(store: Store) -> None:
wrote(store, "2026-09-01", MITHRA, 10)
wrote(store, "2026-09-01", ALA, 10)
wrote(store, "2026-09-01", KUHY, 5)
winners = daily_winners(store.events, GROUP)
assert names_of(winners["2026-09-01"]) == ["Ala"]
@ -35,9 +35,9 @@ def test_the_busiest_person_holds_the_day(store: Store) -> None:
def test_a_tie_shares_the_crown(store: Store) -> None:
"""kuhy's call: shared, because the alternative starts arguments."""
wrote(store, "2026-09-01", MITHRA, 4)
wrote(store, "2026-09-01", ALA, 4)
wrote(store, "2026-09-01", KUHY, 4)
wrote(store, "2026-09-01", IGA, 1)
wrote(store, "2026-09-01", BASIA, 1)
winners = daily_winners(store.events, GROUP)
assert names_of(winners["2026-09-01"]) == ["Ala", "kuhy"]
@ -45,7 +45,7 @@ def test_a_tie_shares_the_crown(store: Store) -> None:
def test_a_tie_keeps_every_tied_run_alive(store: Store) -> None:
"""Three days, both tied throughout: both hold a three-day streak."""
for day in ("2026-09-01", "2026-09-02", "2026-09-03"):
wrote(store, day, MITHRA, 3)
wrote(store, day, ALA, 3)
wrote(store, day, KUHY, 3)
streaks = current_streaks(daily_winners(store.events, GROUP))
assert [(s.display_name, s.length) for s in streaks] == [("Ala", 3), ("kuhy", 3)]
@ -62,9 +62,9 @@ def test_a_run_survives_a_day_nobody_spoke(store: Store) -> None:
def test_losing_a_day_ends_a_run_but_keeps_what_it_reached(store: Store) -> None:
"""The bug this test exists for reported every streak as zero."""
wrote(store, "2026-09-01", MITHRA, 10)
wrote(store, "2026-09-02", MITHRA, 10)
wrote(store, "2026-09-03", MITHRA, 4)
wrote(store, "2026-09-01", ALA, 10)
wrote(store, "2026-09-02", ALA, 10)
wrote(store, "2026-09-03", ALA, 4)
wrote(store, "2026-09-03", KUHY, 4)
wrote(store, "2026-09-06", KUHY, 9)
streaks = current_streaks(daily_winners(store.events, GROUP))
@ -73,45 +73,45 @@ def test_losing_a_day_ends_a_run_but_keeps_what_it_reached(store: Store) -> None
def test_longest_ever_survives_the_run_that_set_it_ending(store: Store) -> None:
wrote(store, "2026-09-01", MITHRA, 10)
wrote(store, "2026-09-02", MITHRA, 10)
wrote(store, "2026-09-03", MITHRA, 10)
wrote(store, "2026-09-01", ALA, 10)
wrote(store, "2026-09-02", ALA, 10)
wrote(store, "2026-09-03", ALA, 10)
wrote(store, "2026-09-04", KUHY, 10)
best = longest_ever(daily_winners(store.events, GROUP))
assert (best.display_name, best.length, best.last_day) == ("Ala", 3, "2026-09-03")
def test_held_days_counts_shared_days_for_everyone_tied(store: Store) -> None:
wrote(store, "2026-09-01", MITHRA, 10)
wrote(store, "2026-09-02", MITHRA, 4)
wrote(store, "2026-09-01", ALA, 10)
wrote(store, "2026-09-02", ALA, 4)
wrote(store, "2026-09-02", KUHY, 4)
winners = daily_winners(store.events, GROUP)
assert held_days(winners, "u-ala") == 2
assert held_days(winners, "u-kuhy") == 1
assert held_days(winners, "u-iga") == 0
assert held_days(winners, "u-basia") == 0
def test_somebody_who_merely_took_part_holds_nothing(store: Store) -> None:
"""Neither leader nor tied: the common case, and the one worth pinning."""
wrote(store, "2026-09-01", MITHRA, 10)
wrote(store, "2026-09-01", ALA, 10)
wrote(store, "2026-09-01", KUHY, 4)
wrote(store, "2026-09-01", IGA, 1)
wrote(store, "2026-09-01", BASIA, 1)
winners = daily_winners(store.events, GROUP)
assert names_of(winners["2026-09-01"]) == ["Ala"]
assert held_days(winners, "u-iga") == 0
assert held_days(winners, "u-basia") == 0
def test_streak_order_is_stable(store: Store) -> None:
"""Longest first, then by name -- not whichever the dict happened to yield."""
wrote(store, "2026-09-01", MITHRA, 3)
wrote(store, "2026-09-01", ALA, 3)
wrote(store, "2026-09-01", KUHY, 3)
wrote(store, "2026-09-01", IGA, 3)
wrote(store, "2026-09-01", BASIA, 3)
streaks = current_streaks(daily_winners(store.events, GROUP))
assert [s.display_name for s in streaks] == ["Basia", "Ala", "kuhy"]
assert [s.display_name for s in streaks] == ["Ala", "Basia", "kuhy"]
def test_the_backfilled_day_can_be_left_out(store: Store) -> None:
wrote(store, "2026-08-30", MITHRA, 98)
wrote(store, "2026-08-30", ALA, 98)
wrote(store, "2026-09-01", KUHY, 5)
winners = daily_winners(store.events, GROUP, exclude="2026-08-30")
assert "2026-08-30" not in winners

73
tests/test_transport_media.py Normal file
View File

@ -0,0 +1,73 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""How an attachment is spelled on the wire.
Split out of test_transport_rest.py, which reached the 250-line cap. The
data-URI form is the container's own, read off its /swagger/doc.json rather
than remembered: api.SendMessageV2 documents base64_attachments as taking
bare base64, "data:<MIME-TYPE>;base64,<DATA>" or
"data:<MIME-TYPE>;filename=<FILENAME>;base64,<DATA>".
"""
from __future__ import annotations
import json
from typing import TYPE_CHECKING
import httpx
from tests.test_transport_rest import MockHandler, _transport
if TYPE_CHECKING:
from signal_bot.config import Config
def _recorder(seen: list[dict[str, object]]) -> MockHandler:
"""A handler that keeps every payload it is sent."""
def record(request: httpx.Request) -> httpx.Response:
seen.append(json.loads(request.content))
return httpx.Response(201)
return record
async def test_an_image_send_is_unchanged_by_the_media_path(config: Config) -> None:
"""send_image delegates now; the payload must still be bare base64."""
seen: list[dict[str, object]] = []
def record(request: httpx.Request) -> httpx.Response:
seen.append(__import__("json").loads(request.content))
return httpx.Response(201)
transport = _transport(config, record)
await transport.send_image("QUJD", "group-a==", caption="hi")
assert seen[0]["base64_attachments"] == ["QUJD"]
assert seen[0]["message"] == "hi"
async def test_a_typed_attachment_travels_as_a_data_uri(config: Config) -> None:
"""The spelling is the container's own, from its /swagger/doc.json."""
seen: list[dict[str, object]] = []
def record(request: httpx.Request) -> httpx.Response:
seen.append(__import__("json").loads(request.content))
return httpx.Response(201)
transport = _transport(config, record)
await transport.send_media("QUJD", "group-a==", media_type="video/mp4", filename="clip.mp4")
assert seen[0]["base64_attachments"] == ["data:video/mp4;filename=clip.mp4;base64,QUJD"]
async def test_a_type_without_a_filename_still_names_the_type(config: Config) -> None:
seen: list[dict[str, object]] = []
def record(request: httpx.Request) -> httpx.Response:
seen.append(__import__("json").loads(request.content))
return httpx.Response(201)
transport = _transport(config, record)
await transport.send_media("QUJD", "group-a==", media_type="image/png")
assert seen[0]["base64_attachments"] == ["data:image/png;base64,QUJD"]

68
tool/e2e_pack.py Executable file
View File

@ -0,0 +1,68 @@
#!/usr/bin/env python3
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Write a synthetic photo pack for the offline end-to-end run.
The real album is private and lives outside this repository, so the e2e
script cannot use it -- and must not need to. This builds a pack of three
invented people, one "both of them" combination and one video, which is
enough to exercise every path the real one uses.
python tool/e2e_pack.py --dir /tmp/pack --group 'group.abc='
"""
from __future__ import annotations
import argparse
import base64
import json
import sys
from pathlib import Path
from typing import Final
# A 1x1 PNG, inline so that no binary file enters the repository.
PIXEL: Final = base64.b64decode(
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg=="
)
# Not playable, and it does not have to be: nothing decodes it, the point is
# that it leaves as video/mp4 rather than as a photo.
CLIP: Final = b"\x00\x00\x00\x18ftypmp42fake"
PEOPLE: Final = [
{"key": "p1", "triggers": ["!alfa"]},
{"key": "p2", "triggers": ["!beta"]},
{"key": "p3", "triggers": ["!gamma"]},
{"key": "p4", "triggers": ["!wideo"]},
]
COMBOS: Final = [{"key": "c1", "triggers": ["!oboje"], "all_of": ["p1", "p2"]}]
TAGS: Final = {
"alfa.png": ["p1"],
"both.png": ["p1", "p2"],
"gamma.png": ["p3"],
"clip.mp4": ["p4"],
}
def write(directory: Path, group: str) -> None:
"""Write the manifest, the tags and the files themselves."""
album = directory / "media"
album.mkdir(parents=True, exist_ok=True)
for name in TAGS:
(album / name).write_bytes(CLIP if name.endswith(".mp4") else PIXEL)
manifest = {"groups": [group], "people": PEOPLE, "combos": COMBOS}
(directory / "people.json").write_text(json.dumps(manifest, indent=2), encoding="utf-8")
(directory / "tags.json").write_text(json.dumps(TAGS, indent=2), encoding="utf-8")
def main() -> int:
"""Build the pack asked for on the command line."""
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--dir", required=True, type=Path, help="where to write the pack")
parser.add_argument("--group", required=True, help="the group id the pack belongs to")
arguments = parser.parse_args()
write(arguments.dir, arguments.group)
sys.stdout.write(f"== wrote a synthetic pack to {arguments.dir}\n")
return 0
if __name__ == "__main__":
raise SystemExit(main())

197
tool/tagger.html Normal file
View File

@ -0,0 +1,197 @@
<!doctype html>
<!-- Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
The tagger's whole interface. Served only on 127.0.0.1, by tool/tagger.py.
No framework and no network calls but its own: this page shows private
photos, and the only thing it may talk to is the process that opened it. -->
<html lang="pl">
<head>
<meta charset="utf-8">
<title>Tagowanie zdjęć</title>
<style>
:root { --bg: #1b1d21; --fg: #f2f3f5; --dim: #9aa0a6; --accent: #7aa2f7; }
* { box-sizing: border-box; }
body { margin: 0; background: var(--bg); color: var(--fg);
font: 16px/1.5 system-ui, sans-serif; }
main { max-width: 900px; margin: 0 auto; padding: 24px; }
h1 { font-size: 20px; margin: 0 0 16px; }
.stage { display: flex; align-items: center; justify-content: center;
height: 62vh; background: #101214; border-radius: 12px; }
.stage img, .stage video { max-width: 100%; max-height: 62vh; border-radius: 12px; }
.people { display: flex; flex-wrap: wrap; gap: 8px; margin: 16px 0; }
.person { border: 1px solid #33363d; border-radius: 999px; padding: 8px 14px;
cursor: pointer; user-select: none; background: #24262b; }
.person.on { background: var(--accent); color: #10131a; border-color: var(--accent); }
.bar { display: flex; align-items: center; justify-content: space-between;
gap: 16px; color: var(--dim); }
button { background: #24262b; color: var(--fg); border: 1px solid #33363d;
border-radius: 8px; padding: 8px 14px; cursor: pointer; }
textarea { width: 100%; min-height: 140px; background: #101214; color: var(--fg);
border: 1px solid #33363d; border-radius: 8px; padding: 12px;
font-family: ui-monospace, monospace; }
.hint { color: var(--dim); font-size: 14px; }
.err { color: #f7768e; }
label.group { display: block; padding: 4px 0; }
</style>
</head>
<body>
<main>
<div id="setup" hidden>
<h1>Kto jest w tej paczce?</h1>
<p class="hint">
Jedna osoba na linię: nazwy komend po przecinku, bez wykrzyknika.<br>
<code>celina, cela</code><br>
Kombinacje po znaku <code>=</code>: <code>oboje = ala + marek</code>
(zdjęcie z obojgiem) albo <code>ktokolwiek = ala | marek</code>
(losowo jedno z nich).
</p>
<textarea id="names" spellcheck="false"></textarea>
<h1>W której grupie działają?</h1>
<div id="groups" class="hint">szukam grup…</div>
<p><button id="save">Zapisz i zacznij tagowanie</button></p>
<p id="error" class="err"></p>
</div>
<div id="tagger" hidden>
<div class="bar">
<h1 id="progress"></h1>
<span class="hint">1–9 = osoba, → dalej, ← wstecz</span>
</div>
<div class="stage" id="stage"></div>
<div class="people" id="people"></div>
<div class="bar">
<button id="prev">← wstecz</button>
<span id="saved" class="hint"></span>
<button id="next">dalej →</button>
</div>
</div>
</main>
<script>
const api = (path, body) => fetch(path, body ? {
method: "POST", headers: {"Content-Type": "application/json"}, body: JSON.stringify(body),
} : undefined).then(async (r) => {
if (!r.ok) { throw new Error((await r.json()).detail || r.statusText); }
return r.json();
});
let state = null;
let at = 0;
function key(name, index) { return name || `p${index + 1}`; }
function parseSetup(text) {
const people = [], combos = [];
const byName = new Map();
for (const raw of text.split("\n").map((l) => l.trim()).filter(Boolean)) {
if (raw.includes("=")) { continue; }
const triggers = raw.split(",").map((t) => "!" + t.trim().replace(/^!/, "")).filter((t) => t.length > 1);
if (!triggers.length) { continue; }
const k = `p${people.length + 1}`;
people.push({key: k, triggers});
byName.set(triggers[0].slice(1), k);
}
let n = 1;
for (const raw of text.split("\n").map((l) => l.trim()).filter(Boolean)) {
if (!raw.includes("=")) { continue; }
const [left, right] = raw.split("=");
const every = right.includes("+");
const parts = right.split(every ? "+" : "|").map((p) => p.trim().replace(/^!/, ""));
const keys = parts.map((p) => byName.get(p)).filter(Boolean);
if (keys.length !== parts.length) { throw new Error(`nie znam kogoś z: ${right.trim()}`); }
const triggers = left.split(",").map((t) => "!" + t.trim().replace(/^!/, ""));
combos.push(every ? {key: `c${n}`, triggers, all_of: keys} : {key: `c${n}`, triggers, any_of: keys});
n += 1;
}
return {people, combos};
}
function drawGroups(known, chosen) {
const box = document.getElementById("groups");
if (!known.length) {
box.innerHTML = '<input id="rawgroup" style="width:100%" placeholder="group.…=">';
return;
}
box.innerHTML = known.map((g, i) => `<label class="group">
<input type="radio" name="group" value="${g.id}" ${chosen.includes(g.id) || (!chosen.length && i === 0) ? "checked" : ""}>
${g.name || "(bez nazwy)"} <span class="hint">${g.id.slice(0, 18)}…</span></label>`).join("");
}
function chosenGroups() {
const raw = document.getElementById("rawgroup");
if (raw) { return [raw.value.trim()]; }
const picked = document.querySelector('input[name="group"]:checked');
return picked ? [picked.value] : [];
}
function entries() {
return [...state.people];
}
function draw() {
const item = state.files[at];
document.getElementById("progress").textContent =
`${at + 1} / ${state.files.length} — ${item.name}`;
const stage = document.getElementById("stage");
stage.innerHTML = item.name.toLowerCase().endsWith(".mp4")
? `<video src="/media/${at}" controls autoplay muted loop></video>`
: `<img src="/media/${at}" alt="">`;
document.getElementById("people").innerHTML = entries().map((p, i) => `
<div class="person ${item.keys.includes(p.key) ? "on" : ""}" data-key="${p.key}">
${i + 1}. ${p.triggers[0].slice(1)}
</div>`).join("");
for (const node of document.querySelectorAll(".person")) {
node.onclick = () => toggle(node.dataset.key);
}
}
async function toggle(k) {
const item = state.files[at];
item.keys = item.keys.includes(k) ? item.keys.filter((x) => x !== k) : [...item.keys, k];
draw();
await api("/api/tag", {file: item.name, keys: item.keys});
document.getElementById("saved").textContent = "zapisano";
setTimeout(() => { document.getElementById("saved").textContent = ""; }, 800);
}
function move(step) {
at = Math.min(state.files.length - 1, Math.max(0, at + step));
draw();
}
document.addEventListener("keydown", (event) => {
if (document.getElementById("tagger").hidden) { return; }
if (event.key === "ArrowRight" || event.key === "Enter") { move(1); }
else if (event.key === "ArrowLeft") { move(-1); }
else if (/^[1-9]$/.test(event.key)) {
const person = entries()[Number(event.key) - 1];
if (person) { toggle(person.key); }
}
});
async function start() {
state = await api("/api/state");
if (state.configured) {
document.getElementById("tagger").hidden = false;
at = Math.max(0, state.files.findIndex((f) => !f.keys.length));
draw();
return;
}
document.getElementById("setup").hidden = false;
drawGroups(state.known_groups, state.groups);
document.getElementById("save").onclick = async () => {
try {
const {people, combos} = parseSetup(document.getElementById("names").value);
if (!people.length) { throw new Error("dodaj przynajmniej jedną osobę"); }
await api("/api/setup", {groups: chosenGroups(), people, combos});
location.reload();
} catch (error) {
document.getElementById("error").textContent = error.message;
}
};
}
document.getElementById("prev").onclick = () => move(-1);
document.getElementById("next").onclick = () => move(1);
start();
</script>
</body>
</html>

221
tool/tagger.py Executable file
View File

@ -0,0 +1,221 @@
#!/usr/bin/env python3
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Tag a private photo album, in a browser, on this machine only.
Sixty-odd photos and seven people is not a hand-edited JSON file. This serves
one page on 127.0.0.1 that shows each item in turn with a numbered checkbox
per person, saves on every keystroke, and never leaves the machine.
Names are typed *into the page*, never as arguments: an argument lands in the
shell history, and the whole design of the pack is that the names exist in
one directory and nowhere else.
python tool/tagger.py # the configured pack
python tool/tagger.py --dir ~/pack --port 8140
python tool/tagger.py --import ~/photos # copy files in, then tag
It prints a summary when it stops: how many items are untagged, and how many
photos each command would have. A command with zero is dead on arrival, and
this is where that shows up rather than in the group.
"""
from __future__ import annotations
import argparse
import json
import shutil
import sys
import webbrowser
from pathlib import Path
from typing import Any, Final
import httpx
import uvicorn
from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse, HTMLResponse
from signal_bot.config_env import load_config
from signal_bot.groups import ConfigError
from signal_bot.pack import MEDIA_DIR, MEDIA_TYPES, PEOPLE_FILE, TAGS_FILE, default_dir
from signal_bot.pack_load import load
PAGE: Final = Path(__file__).with_name("tagger.html")
DEFAULT_PORT: Final = 8140
class Album:
"""The pack directory, and every read and write the page needs."""
def __init__(self, directory: Path) -> None:
"""Point the tagger at one pack, creating the album directory."""
self.directory = directory
(directory / MEDIA_DIR).mkdir(parents=True, exist_ok=True)
@property
def files(self) -> list[str]:
"""Every taggable file, in a stable order."""
album = self.directory / MEDIA_DIR
return sorted(path.name for path in album.iterdir() if path.suffix.lower() in MEDIA_TYPES)
def read(self, name: str) -> dict[str, Any]:
"""One manifest, or an empty one."""
path = self.directory / name
if not path.exists():
return {}
document: dict[str, Any] = json.loads(path.read_text(encoding="utf-8"))
return document
def write(self, name: str, document: dict[str, Any]) -> None:
"""Replace one manifest atomically, so a closed tab loses nothing."""
path = self.directory / name
temporary = path.with_suffix(".tmp")
temporary.write_text(json.dumps(document, indent=2, ensure_ascii=False), encoding="utf-8")
temporary.replace(path)
def state(self) -> dict[str, Any]:
"""Everything the page draws itself from."""
people = self.read(PEOPLE_FILE)
tags = self.read(TAGS_FILE)
return {
"configured": bool(people.get("people")),
"people": people.get("people", []),
"combos": people.get("combos", []),
"groups": people.get("groups", []),
"files": [{"name": name, "keys": tags.get(name, [])} for name in self.files],
}
def setup(
self, groups: list[str], people: list[dict[str, Any]], combos: list[dict[str, Any]]
) -> None:
"""Write people.json, then prove the bot could load it."""
self.write(PEOPLE_FILE, {"groups": groups, "people": people, "combos": combos})
load(self.directory)
def tag(self, filename: str, keys: list[str]) -> None:
"""Record who is in one file."""
if filename not in self.files:
raise HTTPException(status_code=404, detail=f"{filename} is not in the album")
tags = self.read(TAGS_FILE)
tags[filename] = keys
self.write(TAGS_FILE, tags)
def summary(self) -> list[str]:
"""What each command would answer with, and what is still untagged."""
tags = self.read(TAGS_FILE)
lines = [f"== {len(self.files)} items, {len(self.files) - len(tags)} never opened"]
try:
pack = load(self.directory)
except ConfigError as error:
# Reported rather than raised: the summary prints as the tagger
# exits, and a half-built pack is the normal state to stop in.
return [*lines, f"!! the pack does not load yet: {error}"]
for entry in pack.entries:
counts = [len(pack.matching(option)) for option in entry.options]
total = sum(counts)
mark = " " if total else "!!"
lines.append(f"{mark} {entry.triggers[0]}: {total} photo(s)")
return lines
def live_groups() -> list[dict[str, str]]:
"""The groups this account is in, so an id is picked and never typed."""
try:
config = load_config()
response = httpx.get(config.groups_url, timeout=5.0)
response.raise_for_status()
listing = response.json()
except ConfigError, httpx.HTTPError, ValueError:
# No container, no .env, no network: none of them stop the tagging,
# they only mean the group id has to be pasted instead of picked.
return []
return [
{"id": str(entry.get("id", "")), "name": str(entry.get("name", ""))}
for entry in listing
if entry.get("id")
]
def build_app(album: Album) -> FastAPI:
"""The one-page application, bound to one album."""
app = FastAPI()
@app.get("/", response_class=HTMLResponse)
async def page() -> str:
return PAGE.read_text(encoding="utf-8")
@app.get("/api/state")
async def state() -> dict[str, Any]:
return {**album.state(), "known_groups": live_groups()}
@app.post("/api/setup")
async def setup(payload: dict[str, Any]) -> dict[str, str]:
try:
album.setup(
list(payload.get("groups", [])),
list(payload.get("people", [])),
list(payload.get("combos", [])),
)
except ConfigError as error:
# Straight into the page: a manifest the bot would refuse should
# be fixed here, in the form, not found at the next restart.
raise HTTPException(status_code=400, detail=str(error)) from error
return {"status": "ok"}
@app.post("/api/tag")
async def tag(payload: dict[str, Any]) -> dict[str, str]:
album.tag(str(payload["file"]), [str(key) for key in payload.get("keys", [])])
return {"status": "ok"}
@app.get("/media/{index}")
async def media(index: int) -> FileResponse:
# By index, never by a name the page sends back: a filename from a
# client is a path, and portraits.path_for exists for the same reason.
files = album.files
if not 0 <= index < len(files):
raise HTTPException(status_code=404, detail="no such item")
return FileResponse(album.directory / MEDIA_DIR / files[index])
return app
def main() -> int:
"""Serve the tagger until it is stopped, then print the summary."""
parser = argparse.ArgumentParser(
description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
)
parser.add_argument("--dir", type=Path, default=None, help="the pack directory")
parser.add_argument("--port", type=int, default=DEFAULT_PORT)
parser.add_argument(
"--import", dest="source", type=Path, default=None, help="copy media in from here first"
)
parser.add_argument("--no-open", action="store_true", help="do not open a browser")
arguments = parser.parse_args()
directory = arguments.dir or default_dir()[0]
album = Album(directory)
if arguments.source is not None:
copied = _import(arguments.source, directory / MEDIA_DIR)
sys.stdout.write(f"== copied {copied} file(s) into {directory / MEDIA_DIR}\n")
sys.stdout.write(f"== tagging {len(album.files)} item(s) in {directory}\n")
url = f"http://127.0.0.1:{arguments.port}/"
sys.stdout.write(f"== open {url} -- ctrl-c here when you are done\n")
if not arguments.no_open:
webbrowser.open(url)
uvicorn.run(build_app(album), host="127.0.0.1", port=arguments.port, log_level="warning")
sys.stdout.write("\n".join(album.summary()) + "\n")
return 0
def _import(source: Path, album: Path) -> int:
"""Copy, never move: the originals stay put until the tagging is done."""
copied = 0
for path in sorted(source.iterdir()):
if path.suffix.lower() in MEDIA_TYPES and not (album / path.name).exists():
shutil.copy2(path, album / path.name)
copied += 1
return copied
if __name__ == "__main__":
raise SystemExit(main())