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

feat: import a folder of photos without importing it twice

Browse Source 
tool/tagger.py did not run at all: 48957a1 renamed pack.default_dir to
default_dirs and missed the caller, so the tagger raised ImportError at
module scope through two releases. Nothing could have caught it -- tool/ is
outside --cov and mypy ran with [signal_bot, fake_signal]. Both are widened
here, and tests/test_tool_imports.py imports every script under tool/.

--import was a filename check and shutil.copy2. Pointed at a phone export
that overlaps the album it re-imported the same photo under whatever name
the exporter used that time. It now identifies files by magic bytes, strips
them through the same media_strip path !add_photo uses, and dedups on exact
bytes twice: the source hash, then -- only for what survives -- the hash
after stripping. sources.json records both, so a second run on the same
folder is a no-op. What the album refuses is moved to _skipped/<reason>/
beside the source, never deleted.

Two things that made "strip on import" impossible until now:

media_tags claims -all= zeroes the QuickTime date tags and allows them by
value. It does not; measured, it leaves CreateDate at its real value, and
every mp4 was being refused. -time:all= is what zeroes them.

And a tag allowlist cannot see the worst class at all. IMG_6924.mp4 carries
mebx timed-metadata holding per-frame face bounding boxes and a face id,
and those survive a complete tag strip because they live in mdat as opaque
samples. media_tracks.py drops the track with ffmpeg and verifies with
ffprobe that only video and audio survived. This is not the ffmpeg mistake
media_strip warns about: that was ffmpeg replacing exiftool for tags, this
is track removal with exiftool still doing every tag, and the gate is a
positive read-back rather than an exit code.

The roster was write-once. The setup form only appeared on an empty album,
Album.setup overwrote the file whole, and keys were positional (p{len+1}),
so inserting somebody renumbered everyone after them and repointed every
tags.json entry at the wrong person. Keys now come from the highest number
in use, the form merges, and a person can be added mid-session.

--untagged filters the page and never the server's list. /media/{index} and
/api/tag both index the full album, so a shorter list server-side would
show item i of the album while tagging item i of the filtered set.

Measured on the live album: 174 source files -> 1 not media, 82 already in
pack, 24 duplicate in batch, 67 imported, and re-running imports nothing.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WBrE1b9HvFw3xvG9bJiDFt
This commit is contained in:
Krzysztof Rudnicki 2026-08-31 13:22:57 +02:00
parent 11aece2bf7
commit 9d546fef9a
22 changed files with 2184 additions and 128 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
.gitignore vendored
View File

@ -179,6 +179,12 @@ people.json
tags.json
media/
pack/
# sources.json is the import ledger. custom.json and adds.json were never
# listed and hold private data too: custom.json is free text somebody typed
# into the group, adds.json holds Signal member uuids.
sources.json
custom.json
adds.json
# The message database is private and never leaves this machine. A live
# SQLite database is three files, so all three are named.

2
.pre-commit-config.yaml
View File

@ -119,7 +119,7 @@ repos:
entry: .venv/bin/mypy
language: system
pass_filenames: false
args: [signal_bot, fake_signal]
args: [signal_bot, fake_signal, tool]
- id: shellcheck
name: shellcheck

53
DOCS-add-photo.md
View File

@ -43,10 +43,17 @@ it. Do not reintroduce ffmpeg for this.
So the gate is a **positive check on the output**, never an exit code:
```sh
exiftool -all= -o <dst> <src> # never --overwrite_original on the temp
exiftool -G0 -s -all <dst> # then read it back
exiftool -all= -time:all= -o <dst> <src> # never --overwrite_original
exiftool -G0 -s -all <dst> # then read it back
```
`-time:all=` is **not** redundant with `-all=`. `media_tags.ZEROED_DATE_TAGS`
allows the QuickTime date tags "by value" on the grounds that `-all=` zeroes
rather than deletes them. Measured 2026-08-31: it does not. `-all=` alone left
`QuickTime:CreateDate` at its real value, and the consequence was that **every
mp4 was refused** — the code documented behaviour it did not get. `-time:all=`
is what actually zeroes them.
and every surviving tag must be **structural**. Measured against exiftool
13.55, the plan's original "the surviving set must be empty" would have
rejected every format except JPEG — a valid PNG always reports its IHDR under
@ -59,6 +66,48 @@ is refused.
Allowlist, not blocklist, for the reason a naive gate misses: an MP4 geotag is
a QuickTime `udta`/`©xyz` atom reported under **`QuickTime`**, not `GPS`.
### Tracks: ffmpeg, for what no tag allowlist can see
A tag allowlist is structurally blind to one class, and it is the worst one.
`IMG_6924.mp4` in the live album carries two `mebx` timed-metadata tracks, and
after a *complete* `-all= -time:all=` pass its bytes still contain:
```
com.apple.quicktime.detected-face
com.apple.quicktime.detected-face.bounds
com.apple.quicktime.detected-face.face-id
com.apple.quicktime.detected-face.roll-angle / .yaw-angle
```
Per-frame face bounding boxes and a face identity number. exiftool is not
broken for missing them: `mebx` samples live in `mdat` as opaque payload, not
as tags, so **there is no tag to refuse**. The only answer is to drop the
track, which `signal_bot/media_tracks.py` does before exiftool runs:
```sh
ffmpeg -v error -i <src> -map 0:v -map "0:a?" -c copy \
-map_metadata -1 -map_chapters -1 -y <dst>
ffprobe -v error -show_entries stream=codec_type -of csv=p=0 <dst>
```
and every surviving stream must be `video` or `audio`.
**This is not the ffmpeg mistake above.** That one was ffmpeg replacing
exiftool for *tag* removal, where it silently failed on JPEG. Here ffmpeg
removes *tracks*, exiftool still does every tag removal afterwards, and the
gate is a positive read-back of which streams survived rather than an exit
code. `-c copy` is safe for this job precisely because nothing is being
claimed about the frames — the claim is about which tracks exist, and ffprobe
answers that directly. Do not collapse the two jobs back together.
`QuickTime:MetaFormat` is deliberately **absent** from the allowlist: a file
still reporting one has a metadata track the re-mux should have removed, and
refusing it is the correct outcome.
**Both entry points obey this.** `tool/tagger.py --import` used to be a
verbatim `shutil.copy2`; since 2026-08-31 it runs the same stripper, so the
guarantee on this page covers the whole album and not just the chat path.
The list includes the **codec's own configuration** — `CompressorName`,
`VendorID`, and the HEVC parameter set (`HEVCConfigurationVersion`,
`GeneralProfileIDC`, `BitDepthLuma`, …). That was forced by measurement, not

81
DOCS-photo-pack.md
View File

@ -139,6 +139,34 @@ gap in its history; the adder has to be recorded at add time or not at all.
It lives inside the album, which is `0700` and outside this repository, so it
is handled exactly like the photos.
## sources.json
```json
{
"IMG_0234.jpg": {"source": "9f2c…", "stored": "b71e…"},
"a3ae52c7….jpg": {"source": null, "stored": "a3ae…"}
}
```
The import ledger: every file the album has ever taken in, under **two**
SHA-256s, because stripping changes the bytes.
- `source` — the file as it *arrived*, before metadata was removed. This
answers "have I imported this photo before?" when the same export folder is
pointed at the album a second time.
- `stored` — the bytes on disk now. This answers "is the photo somebody just
uploaded from the chat already here?", because `!add_photo` hashes *after*
stripping.
Neither alone is enough. The same photo arriving through `--import` and
through Signal has different source bytes — the phone re-encodes — but strips
to the same output. `source` is `null` when the original bytes were never
kept, which is exactly the `!add_photo` files.
Like `adds.json`, **`pack_load` never touches it.** A corrupt ledger degrades
to "I have seen nothing", costing a duplicate photo; a corrupt ledger checked
at startup would be a restart loop under `Restart=always`.
## Adding from the chat: !add_photo
A member of one of the album's groups can send a photo captioned
@ -150,15 +178,54 @@ the album format itself: **`DOCS-add-photo.md`**.
## 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
python tool/tagger.py --import ~/photos # dedup, strip, import, tag
python tool/tagger.py --import ~/photos --dry-run # the report only
python tool/tagger.py --untagged # only items with no tags
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.
It serves one page on `127.0.0.1:8140`: the item, a labelled button per
person, `1`–`0` then `q`–`p` 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.
**`--import` is not a copy.** Every file is identified by its magic bytes
(not its suffix), stripped through the same `media_strip` path `!add_photo`
uses, and checked against `sources.json` twice: the source hash first, then
the post-strip hash. What the album refuses is **moved, never deleted**, to
`_skipped/<reason>/` beside the source folder, and the printed report has to
account for every file seen:
```
174 files seen
1 not media -> _skipped/not-media/
0 could not be stripped -> _skipped/unstrippable/
82 already in pack -> _skipped/already-in-pack/
24 duplicate in batch -> _skipped/exact-dup/
67 imported
---
174 accounted for 85 -> 152 in media/
```
Every bucket prints even at zero: a suppressed zero is a silently missing
category, and `0 could not be stripped` is how you see the video pipeline is
still working. Re-running on the same folder is a no-op — that is what the
ledger is for.
**`--untagged` filters the page, never the server's list.** `/media/{index}`
and `/api/tag` both index the full album, so a shorter list on the server
would show item *i* of the album while tagging item *i* of the filtered set —
every photo tagged with the wrong people, and no error anywhere.
**Adding a person works mid-session.** The roster used to be write-once: the
setup form only appeared on an empty album, `Album.setup` overwrote the file
whole, and keys were positional (`p{len+1}`), so inserting somebody renumbered
everyone after them and repointed every `tags.json` entry at the wrong person.
Now keys come from the highest number in use, the form merges, and there is an
"add person" box in the tagger itself — for the photo of somebody new that is
already on the screen.
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**;

20
scripts/install_service.sh
View File

@ -68,9 +68,29 @@ ensure_exiftool() {
fi
}
# ffmpeg drops the metadata *tracks* exiftool cannot see. An iPhone clip
# carries mebx timed-metadata holding per-frame face bounding boxes and a
# face id, and those survive a complete "-all= -time:all=" pass because they
# live in mdat as opaque samples, not as tags. Without ffmpeg every video is
# refused rather than stored with them.
ensure_ffmpeg() {
if ! command -v ffmpeg >/dev/null || ! command -v ffprobe >/dev/null; then
echo "Installing ffmpeg (needed to store video)..."
if ! sudo pacman -S --needed --noconfirm ffmpeg; then
die "could not install ffmpeg; run: sudo pacman -S ffmpeg"
fi
fi
# Same trap as exiftool: ask the bot's own resolver, under the unit's PATH.
if ! env -i "PATH=/usr/local/bin:/usr/bin" "${REPO_ROOT}/.venv/bin/python" -c \
'import sys; from signal_bot.media_tracks import locate; sys.exit(0 if locate("ffmpeg") and locate("ffprobe") else 1)'; then
die "ffmpeg is installed but signal_bot.media_tracks.locate() cannot find it under the unit's PATH"
fi
}
validate_requirements() {
command -v systemctl >/dev/null || die "systemctl not found; this needs systemd"
ensure_exiftool
ensure_ffmpeg
[[ -f "${REPO_ROOT}/.env" ]] ||
die "no ${REPO_ROOT}/.env -- run scripts/register_bot.sh first"
[[ -x "${REPO_ROOT}/run.sh" ]] || die "no executable ${REPO_ROOT}/run.sh"

18
signal_bot/media_strip.py
View File

@ -35,6 +35,7 @@ from typing import TYPE_CHECKING, Final, Protocol
from signal_bot.external import run
from signal_bot.media_tags import is_allowed
from signal_bot.media_tracks import ISO_SUFFIXES, flatten
if TYPE_CHECKING:
from signal_bot.media_sniff import MediaKind
@ -129,10 +130,25 @@ class ExiftoolStripper:
"""Do the strip inside a temp directory that is removed either way."""
source = workspace / f"in{kind.suffix}"
destination = workspace / f"out{kind.suffix}"
if kind.suffix in ISO_SUFFIXES:
# Tracks first, tags second. exiftool cannot see a mebx track at
# all -- see media_tracks for the face-detection payload that
# survives a complete tag strip -- so this has to happen here or
# not at all.
flattened = flatten(payload, kind.suffix)
if flattened is None:
return None
payload = flattened
source.write_bytes(payload)
# Never --overwrite_original: writing in place would leave a partial
# file where the original was if exiftool died halfway.
if not self._exiftool_ok(["-all=", "-o", str(destination), str(source)]):
#
# "-time:all=" is not redundant with "-all=". media_tags says the date
# tags are "zeroed, not deleted" by -all= and allows them by value;
# measured on 2026-08-31, -all= alone leaves QuickTime:CreateDate at
# its real value and every mp4 in the album was refused because of it.
# -time:all= is what actually zeroes them.
if not self._exiftool_ok(["-all=", "-time:all=", "-o", str(destination), str(source)]):
return None
if not destination.is_file():
logger.warning("exiftool produced no output for %s", kind.suffix)

8
signal_bot/media_tags.py
View File

@ -61,6 +61,9 @@ _PNG: Final = frozenset(
"PixelsPerUnitX",
"PixelsPerUnitY",
"PixelUnits",
# The sBIT chunk, "8 8 8 8". Says how many bits of each channel
# are significant; carries nothing about who or where.
"SignificantBits",
}
)
@ -112,6 +115,11 @@ _RIFF: Final = frozenset(
# scripts/e2e.sh still proves a QuickTime GPS atom does not survive.
_QUICKTIME: Final = frozenset(
{
# Deliberately absent: "MetaFormat". A file still reporting one
# has a metadata track that media_tracks.flatten should have
# removed, and refusing it is the correct outcome.
"ExtendedLanguageTag",
"ContentDescribes",
"MajorBrand",
"MinorVersion",
"CompatibleBrands",

151
signal_bot/media_tracks.py Normal file
View File

@ -0,0 +1,151 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Dropping the tracks a video carries besides its picture and its sound.
An iPhone video is not one stream. Measured on a clip already sitting in the
album, ``IMG_6924.mp4`` carries two ``mebx`` timed-metadata tracks, and after a
*complete* ``exiftool -all= -time:all=`` pass its bytes still contain::
com.apple.quicktime.detected - face
com.apple.quicktime.detected - face.bounds
com.apple.quicktime.detected - face.face - id
com.apple.quicktime.detected - face.roll - angle
com.apple.quicktime.detected - face.yaw - angle
Per-frame face bounding boxes and a face identity number -- biometric data
about the people in a private album. exiftool cannot remove it and is not
broken for failing to: ``mebx`` samples live in ``mdat`` as opaque payload, not
in a tag. A tag allowlist is structurally incapable of seeing this class, which
is why removing the whole track is the only answer.
**This is not the ffmpeg mistake media_strip.py warns about.** That warning is
about using ffmpeg to remove *tags*, where ``-map_metadata -1 -c copy`` returned
a byte-identical JPEG with its GPS marker intact and exited 0. Here ffmpeg
removes *tracks*, exiftool still does every tag removal afterwards, and the
check below is a positive read-back of which streams survived rather than an
exit code. Do not collapse the two jobs back together.
``-c copy`` is safe for this job precisely because nothing is being claimed
about the frames: the assertion is about which tracks exist, and ffprobe
answers that directly.
"""
from __future__ import annotations
import logging
import shutil
import tempfile
from pathlib import Path
from typing import Final
from signal_bot.external import run
logger = logging.getLogger(__name__)
FFMPEG: Final = "ffmpeg"
FFPROBE: Final = "ffprobe"
# The containers that can carry a metadata track at all. Everything else the
# album accepts is a still image and goes straight past this module.
ISO_SUFFIXES: Final = frozenset({".mp4", ".mov", ".m4v"})
# The only stream kinds a stored file may have once it is flattened.
KEPT: Final = frozenset({"video", "audio"})
# Same reasoning as media_strip.FALLBACK_PATHS: a systemd unit's PATH is
# "/usr/local/bin:/usr/bin" and nothing else.
FALLBACK_DIRS: Final = ("/usr/local/bin", "/usr/bin", "/opt/homebrew/bin")
TIMEOUT_SECONDS: Final = 120
def locate(binary: str) -> str | None:
"""One tool's absolute path, or None when it is not installed."""
found = shutil.which(binary)
if found is not None:
return found
candidates = (Path(directory) / binary for directory in FALLBACK_DIRS)
return next((str(path) for path in candidates if path.is_file()), None)
def flatten(payload: bytes, suffix: str) -> bytes | None:
"""Re-mux one video keeping only its picture and sound.
Args:
payload: The file exactly as it arrived.
suffix: The sniffed suffix, which decides the container ffmpeg writes.
Returns:
The re-muxed file, or None when either tool is missing, ffmpeg fails,
or the result still reports a stream that is not video or audio.
"""
ffmpeg = locate(FFMPEG)
ffprobe = locate(FFPROBE)
if ffmpeg is None or ffprobe is None:
logger.error("ffmpeg/ffprobe not found; refusing to store %s", suffix)
return None
with tempfile.TemporaryDirectory(prefix="signal-bot-tracks-") as workspace:
return _run(Path(workspace), payload, suffix, ffmpeg, ffprobe)
def _run(workspace: Path, payload: bytes, suffix: str, ffmpeg: str, ffprobe: str) -> bytes | None:
"""Do the re-mux inside a temp directory that is removed either way."""
source = workspace / f"in{suffix}"
destination = workspace / f"out{suffix}"
source.write_bytes(payload)
result = run(
[
ffmpeg,
"-v",
"error",
"-i",
str(source),
# "0:a?" and not "0:a": a clip with no sound is still a clip, and
# a hard mapping would refuse it.
"-map",
"0:v",
"-map",
"0:a?",
"-c",
"copy",
"-map_metadata",
"-1",
"-map_chapters",
"-1",
"-y",
str(destination),
],
timeout=TIMEOUT_SECONDS,
)
if not result.ok or not destination.is_file():
logger.error("ffmpeg could not re-mux %s: %s", suffix, result.stderr.strip())
return None
kinds = streams(destination, ffprobe)
if kinds is None or not kinds or not set(kinds) <= KEPT:
logger.error("refusing %s: streams after re-mux were %s", suffix, kinds)
return None
return destination.read_bytes()
def streams(path: Path, ffprobe: str) -> tuple[str, ...] | None:
"""Every stream's kind, in order, or None when they cannot be read.
The positive check this module rests on. An empty result is treated as a
failure by the caller rather than as "no bad streams": a file ffprobe
cannot describe is one nobody can vouch for.
"""
result = run(
[
ffprobe,
"-v",
"error",
"-show_entries",
"stream=codec_type",
"-of",
"csv=p=0",
str(path),
],
timeout=TIMEOUT_SECONDS,
)
if not result.ok:
return None
return tuple(line.strip().rstrip(",") for line in result.stdout.splitlines() if line.strip())

122
signal_bot/pack_album.py Normal file
View File

@ -0,0 +1,122 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""The album as the tagging page sees it: every read and write it needs.
Split out of ``tool/tagger.py`` when the importer landed and the two together
went past the shared 250-line cap. It lives here rather than beside the tagger
because this is the half worth testing -- ``tool/`` is outside the coverage
gate, which is exactly how an unconditional ImportError sat in ``tagger.py``
through two releases. What is left there is FastAPI decorators and
``uvicorn.run``.
Nothing here raises an HTTP exception. The routes translate; this file deals
in ``KeyError`` and ``ConfigError``, so it can be tested without a client.
"""
from __future__ import annotations
import json
from typing import TYPE_CHECKING, Any
from signal_bot.groups import ConfigError
from signal_bot.pack import MEDIA_DIR, MEDIA_TYPES, PEOPLE_FILE, TAGS_FILE
from signal_bot.pack_load import load
from signal_bot.pack_people import add_person, merge
if TYPE_CHECKING:
from pathlib import Path
class Album:
"""The pack directory, and every read and write the page needs."""
def __init__(self, directory: Path, *, untagged_only: bool = False) -> None:
"""Point the tagger at one pack, creating the album directory."""
self.directory = directory
self.untagged_only = untagged_only
(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")),
# A view filter, never a different list: /media/{index} and
# /api/tag both index self.files, so filtering server-side
# would show item i of the album while tagging item i of the
# filtered set -- every photo tagged with the wrong people.
"untagged_only": self.untagged_only,
"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:
"""Merge into people.json, then prove the bot could load it.
Merged rather than replaced: this form used to overwrite the file
and hand out keys by position, so re-running it on a configured
album renumbered everyone and repointed every tag.
"""
document = self.read(PEOPLE_FILE)
self.write(
PEOPLE_FILE,
{"groups": groups, "people": merge(document, people), "combos": combos},
)
load(self.directory)
def add_person(self, triggers: list[str]) -> str:
"""Append one person to the roster, mid-session, and return the key."""
return add_person(self.directory, triggers)
def tag(self, filename: str, keys: list[str]) -> None:
"""Record who is in one file."""
if filename not in self.files:
msg = f"{filename} is not in the album"
raise KeyError(msg)
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:
# Distinct files, not the sum per option: a photo showing two of
# the three people an "either of them" entry covers is one photo
# the command can return, and summing would claim it twice.
reachable = {name for option in entry.options for name in pack.matching(option)}
mark = " " if reachable else "!!"
lines.append(f"{mark} {entry.triggers[0]}: {len(reachable)} photo(s)")
return lines

245
signal_bot/pack_import.py Normal file
View File

@ -0,0 +1,245 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Importing a folder of photos into the album without importing it twice.
``tool/tagger.py --import`` used to be a filename check and ``shutil.copy2``.
Pointed at a phone export that overlaps what is already tagged, that copies
the same photo in again under whatever name the exporter gave it that time,
and copies the metadata in with it. Both are fixed here:
* **What a file is comes from its bytes** (media_sniff), not its suffix, the
same rule ``!add_photo`` already follows. A suffix that disagrees with the
bytes is replaced, because ``pack.media_type`` looks the stored suffix up in
a closed table and a miss there fails the *next* load, not this import.
* **Metadata is stripped on the way in**, through the same media_strip
stripper the chat path uses, so the album has one rule instead of two.
Dedup is exact bytes, twice, in this order: the source hash against everything
the ledger has ever taken in, then -- only for what survives -- the stripped
hash against what is on disk. The second stage is what makes a photo uploaded
from the chat and the same photo in an export folder collapse into one, even
though their source bytes differ.
Whether a repeat is called "already in pack" or "duplicate in batch" is decided
by *when* the hash was first seen: in the ledger before the run started, or
earlier in this same run. That is the only difference between the two, and it is
what lets the report's arithmetic be read as a funnel.
Nothing is deleted. A file the album refuses is *moved* to a folder beside the
source, one directory per reason, so a wrong call is visible and reversible
rather than gone.
"""
from __future__ import annotations
import shutil
from dataclasses import dataclass, field
from typing import TYPE_CHECKING
from signal_bot import pack_ledger
from signal_bot.media_sniff import sniff
from signal_bot.pack import MEDIA_DIR, MEDIA_TYPES
from signal_bot.pack_custom import FILE_MODE
from signal_bot.pack_ledger import Entry
from signal_bot.pack_report import SKIPPED_DIR, ImportReport, Outcome
if TYPE_CHECKING:
from collections.abc import Iterator
from pathlib import Path
from signal_bot.media_sniff import MediaKind
from signal_bot.media_strip import Stripper
@dataclass(frozen=True, slots=True)
class Stored:
"""Everything an accepted file needs, computed once during planning.
Carried so the apply pass never strips a second time: exiftool is the
slow part of an import, and stripping twice would also mean trusting it
to be byte-deterministic, which is exactly what the ledger exists to
avoid depending on.
"""
name: str
payload: bytes
entry: Entry
@dataclass(frozen=True, slots=True)
class Decision:
"""What one source file is, and what the album will do with it."""
source: Path
outcome: Outcome
# Set for IMPORTED and for nothing else, which is what the apply pass
# branches on -- so there is no state where the two disagree.
stored: Stored | None = None
@dataclass
class _Seen:
"""Hashes and names already spoken for, split by when they were seen."""
ledger_source: set[str]
ledger_stored: set[str]
taken: set[str]
batch_source: set[str] = field(default_factory=set)
batch_stored: set[str] = field(default_factory=set)
def plan_import(source: Path, directory: Path, stripper: Stripper) -> ImportReport:
"""Decide what would happen, without storing or moving anything.
Not quite side-effect free: it backfills the ledger, which records what
the album already holds. No photo is written, moved or altered.
Args:
source: The folder to import from.
directory: The album directory.
stripper: How metadata is removed; injected so tests never shell out.
Returns:
A report whose buckets account for every file in ``source``.
"""
# Backfilled here rather than behind a flag somebody has to remember:
# an empty ledger makes every photo already in the album look new, and
# the failure is a report that says "173 imported" instead of "85".
entries = pack_ledger.backfill(directory)
album = directory / MEDIA_DIR
seen = _Seen(
ledger_source=pack_ledger.source_hashes(entries),
ledger_stored=pack_ledger.stored_hashes(entries),
taken={path.name for path in album.iterdir()} if album.is_dir() else set(),
)
before = len(seen.taken)
decisions = tuple(_decide(path, stripper, seen) for path in _candidates(source))
return ImportReport(source=source, directory=directory, decisions=decisions, before=before)
def run_import(source: Path, directory: Path, stripper: Stripper) -> ImportReport:
"""Import a folder: store what is new, move aside what is not.
Args:
source: The folder to import from.
directory: The album directory.
stripper: How metadata is removed.
Returns:
The report describing what was just done. A decision the recheck
below overturns is reported as ALREADY_IN_PACK, not as imported.
"""
report = plan_import(source, directory, stripper)
album = directory / MEDIA_DIR
album.mkdir(parents=True, exist_ok=True)
# Re-read rather than reuse the planning copy: planning backfilled, and a
# photo may have arrived from the chat in between.
entries = pack_ledger.backfill(directory)
known = pack_ledger.stored_hashes(entries)
settled: list[Decision] = []
for decision in report.decisions:
if decision.stored is None:
_reject(source, decision)
settled.append(decision)
continue
if decision.stored.entry.stored in known:
# Planning decided this was new; somebody uploaded it from the
# chat while the strip pass was still running. Measured on
# 2026-08-31: 6 photos landed that way during one import, and
# storing them anyway put each in the album twice -- once
# hash-named by !add_photo and once under its original name.
_reject(source, Decision(decision.source, Outcome.ALREADY_IN_PACK))
settled.append(Decision(decision.source, Outcome.ALREADY_IN_PACK))
continue
_store(album, decision.stored)
entries[decision.stored.name] = decision.stored.entry
known.add(decision.stored.entry.stored)
settled.append(decision)
pack_ledger.write(directory, entries)
return ImportReport(
source=report.source,
directory=report.directory,
decisions=tuple(settled),
before=report.before,
)
def _candidates(source: Path) -> Iterator[Path]:
"""Every regular file in the source folder, in a stable order."""
return (path for path in sorted(source.iterdir()) if path.is_file())
def _decide(path: Path, stripper: Stripper, seen: _Seen) -> Decision:
"""Bucket one file, claiming its hashes and name in ``seen`` as it goes."""
payload = path.read_bytes()
kind = sniff(payload)
if kind is None:
return Decision(path, Outcome.NOT_MEDIA)
source_hash = pack_ledger.digest(payload)
repeat = _repeat(source_hash, seen.batch_source, seen.ledger_source)
if repeat is not None:
return Decision(path, repeat)
cleaned = stripper.strip(payload, kind)
if cleaned is None:
return Decision(path, Outcome.UNSTRIPPABLE)
stored_hash = pack_ledger.digest(cleaned)
seen.batch_source.add(source_hash)
repeat = _repeat(stored_hash, seen.batch_stored, seen.ledger_stored)
if repeat is not None:
return Decision(path, repeat)
seen.batch_stored.add(stored_hash)
name = _stored_name(path, kind, seen.taken)
seen.taken.add(name)
return Decision(
path,
Outcome.IMPORTED,
stored=Stored(
name=name, payload=cleaned, entry=Entry(source=source_hash, stored=stored_hash)
),
)
def _repeat(digest: str, batch: set[str], ledger: set[str]) -> Outcome | None:
"""Which kind of repeat this hash is, or None when it is new.
The only thing separating the two is *when* the hash was first seen: in
the ledger before the run started, or earlier in this same run.
"""
if digest in batch:
return Outcome.EXACT_DUP
if digest in ledger:
return Outcome.ALREADY_IN_PACK
return None
def _stored_name(path: Path, kind: MediaKind, taken: set[str]) -> str:
"""The name this file takes in media/: original, made safe and unique.
The suffix is replaced when it disagrees with the bytes, because
``pack.media_type`` looks it up in a closed table and a miss there fails
the next load. A name already in use by *different* bytes is
disambiguated rather than skipped -- skipping is the old bug, and it lost
a photo silently every time two exports both said "IMG_0001.jpg".
"""
suffix = path.suffix if MEDIA_TYPES.get(path.suffix.lower()) == kind.media_type else kind.suffix
candidate = f"{path.stem}{suffix}"
counter = 1
while candidate in taken:
counter += 1
candidate = f"{path.stem}-{counter}{suffix}"
return candidate
def _store(album: Path, stored: Stored) -> None:
"""Write one accepted file into media/, atomically."""
path = album / stored.name
temporary = path.with_name(f"{stored.name}.tmp")
temporary.write_bytes(stored.payload)
temporary.chmod(FILE_MODE)
temporary.replace(path)
def _reject(source: Path, decision: Decision) -> None:
"""Move one refused file into a folder beside the source, by reason."""
target = source.parent / SKIPPED_DIR / str(decision.outcome)
target.mkdir(parents=True, exist_ok=True)
shutil.move(str(decision.source), str(target / decision.source.name))

160
signal_bot/pack_ledger.py Normal file
View File

@ -0,0 +1,160 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""What the album has already seen, so importing the same folder twice is free.
Two hashes per stored file, because stripping changes the bytes:
* ``source`` -- SHA-256 of the file as it arrived, *before* metadata was
removed. This is the one that answers "have I imported this photo before?"
when the same export folder is pointed at the album a second time.
* ``stored`` -- SHA-256 of the bytes on disk now. This is the one that answers
"is the photo somebody just uploaded from the chat already here?", because
``!add_photo`` hashes *after* stripping (pack_store.content_name).
Neither alone is enough. A photo that arrived through ``--import`` and the same
photo re-sent through Signal have different source bytes -- the phone re-encodes
-- but strip to the same output; a photo re-imported from the same folder has
identical source bytes but restripping is not guaranteed byte-for-byte across
exiftool versions. Recording both makes each case a lookup.
``source`` is None when the original bytes are unrecoverable, which is exactly
the ``!add_photo`` files: their stored name is a hash of the *stripped* payload
and the upload was never kept.
Unlike pack_load, nothing here ever raises on a bad file. This ledger is an
optimisation and a safety net, not a manifest: ``pack_load`` failing at startup
is a restart loop under ``Restart=always`` (see pack_holder.reload, which
deliberately never raises), and a corrupt ledger degrading to "I have seen
nothing" costs a duplicate photo, while a corrupt ledger taking startup down
costs the bot.
"""
from __future__ import annotations
import hashlib
import json
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Final
from signal_bot.pack import MEDIA_DIR
from signal_bot.pack_store import ADDS_FILE, write_json
if TYPE_CHECKING:
from collections.abc import Mapping
from pathlib import Path
LEDGER_FILE: Final = "sources.json"
@dataclass(frozen=True, slots=True)
class Entry:
"""The two hashes one stored file is known by."""
# The bytes as they arrived, or None when they were never kept.
source: str | None
# The bytes on disk now.
stored: str
def digest(payload: bytes) -> str:
"""The hash this module identifies bytes by."""
return hashlib.sha256(payload).hexdigest()
def read(directory: Path) -> dict[str, Entry]:
"""The ledger, or an empty one for anything unreadable or malformed.
Args:
directory: The album directory.
Returns:
Stored filename to its entry. Malformed individual entries are
dropped rather than failing the whole read: a hand-edited file
should cost the entries somebody broke, not all of them.
"""
entries: dict[str, Entry] = {}
for name, value in _document(directory / LEDGER_FILE).items():
entry = _entry(value)
if entry is not None:
entries[name] = entry
return entries
def write(directory: Path, entries: Mapping[str, Entry]) -> None:
"""Replace the ledger atomically, 0600, the way every manifest is written."""
document = {
name: {"source": entry.source, "stored": entry.stored}
for name, entry in sorted(entries.items())
}
write_json(directory / LEDGER_FILE, document)
def source_hashes(entries: Mapping[str, Entry]) -> set[str]:
"""Every known pre-strip hash, for "did this file arrive here before?"."""
return {entry.source for entry in entries.values() if entry.source is not None}
def stored_hashes(entries: Mapping[str, Entry]) -> set[str]:
"""Every known on-disk hash, for "is this photo already in the album?"."""
return {entry.stored for entry in entries.values()}
def backfill(directory: Path) -> dict[str, Entry]:
"""Record every file already in media/ that the ledger does not know.
Run this **before** any retroactive strip. The files that arrived through
``tool/tagger.py --import`` were copied verbatim (shutil.copy2), so their
current bytes *are* their source bytes -- the one and only moment that
fact is recoverable. Strip first and it is gone, and re-importing the
same export folder next year silently re-adds all of them.
The ``!add_photo`` files are told apart by ``adds.json``, which is the
exact record of what came in through the chat. Those were stripped on the
way in, so their source bytes never existed on disk and stay None.
Args:
directory: The album directory.
Returns:
The merged ledger, already written.
"""
entries = read(directory)
from_chat = set(_document(directory / ADDS_FILE))
album = directory / MEDIA_DIR
for path in sorted(album.iterdir()) if album.is_dir() else ():
if not path.is_file() or path.name in entries:
continue
stored = digest(path.read_bytes())
entries[path.name] = Entry(
source=None if path.name in from_chat else stored,
stored=stored,
)
write(directory, entries)
return entries
def _entry(value: object) -> Entry | None:
"""One ledger row, or None when it is not shaped like one."""
if not isinstance(value, dict):
return None
stored = value.get("stored")
source = value.get("source")
if not isinstance(stored, str) or not isinstance(source, str | None):
return None
return Entry(source=source, stored=stored)
def _document(path: Path) -> dict[str, Any]:
"""One JSON object, or an empty one for anything unreadable.
Forgiving on purpose -- see the module docstring. This is the same policy
pack_store._document has for adds.json, and the opposite of pack_load's.
"""
try:
raw = path.read_bytes()
except OSError:
return {}
try:
document = json.loads(raw)
except json.JSONDecodeError:
return {}
return document if isinstance(document, dict) else {}

183
signal_bot/pack_people.py Normal file
View File

@ -0,0 +1,183 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Adding somebody to the roster of an album that already has one.
The tagger could only ever write ``people.json`` from empty: the setup form
appeared when there were no people, ``Album.setup`` overwrote the file whole,
and the keys were handed out by position -- ``p{len + 1}``. Together those
mean the obvious thing (edit the list to insert somebody) renumbers everyone
after them, and every ``tags.json`` entry then points at the wrong person,
silently, with no error at any layer. So in practice a roster was written once
and never touched, and a new face meant hand-editing JSON.
This module is the append-only path instead. Keys come from the highest number
in use, never the count -- the same rule ``pack_custom.next_key`` follows, and
for the same reason: deleting ``p3`` by hand must not let a later add reuse the
key and inherit its photos.
Two checks before anything is written, and they are different in kind:
* **Against the album**, by writing the file and then calling ``pack_load.load``
on it. That is the whole of pack_load's validation -- trigger shape, no
digits, uniqueness after folding, across people and combos and custom labels
-- rather than a second copy of it here that could drift. A failure rolls the
file back to the bytes it had.
* **Against the bot's own commands**, from ``aliases.json``. ``load`` cannot see
those: a person triggered by ``!losuj`` is a perfectly valid pack that
quietly shadows ``!random`` at run time. This is the same collision
``pack_add`` refuses a custom label for.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import TYPE_CHECKING, Any, Final
from signal_bot.folding import fold
from signal_bot.groups import ConfigError
from signal_bot.pack import PEOPLE_FILE
from signal_bot.pack_load import load
from signal_bot.pack_store import write_json
if TYPE_CHECKING:
from collections.abc import Sequence
ALIAS_FILE: Final = Path(__file__).with_name("aliases.json")
# The person-key shape, mirroring pack_custom.KEY_PATTERN for custom tags.
KEY_PREFIX: Final = "p"
def reserved() -> frozenset[str]:
"""Every folded word that is already one of the bot's own commands."""
document = json.loads(ALIAS_FILE.read_text(encoding="utf-8"))
return frozenset(
fold(trigger)
for name, triggers in document.items()
if not name.startswith("_")
for trigger in triggers
)
def next_key(people: Sequence[dict[str, Any]]) -> str:
"""The key a new person would take, without taking it.
One past the highest number in use rather than the count, so a roster
somebody has edited by hand cannot make an add reuse a retired key.
"""
used = [
int(key.removeprefix(KEY_PREFIX))
for person in people
if isinstance(key := person.get("key"), str)
and key.startswith(KEY_PREFIX)
and key.removeprefix(KEY_PREFIX).isdigit()
]
return f"{KEY_PREFIX}{max(used, default=0) + 1}"
def normalise(triggers: Sequence[str]) -> list[str]:
"""Trigger spellings as they are stored: "!"-prefixed, blanks dropped.
People type "alfa" as often as "!alfa" and the difference is not one
worth an error message.
"""
cleaned = []
for raw in triggers:
word = raw.strip()
if word:
cleaned.append(word if word.startswith("!") else f"!{word}")
return cleaned
def add_person(directory: Path, triggers: Sequence[str]) -> str:
"""Append one person to the roster and prove the album still loads.
Args:
directory: The album directory.
triggers: What somebody types to get this person's photos.
Returns:
The key the new person was given.
Raises:
ConfigError: The triggers are empty, collide with one of the bot's
own commands, or leave the album in a state that does not load.
The manifest is unchanged in every one of those cases.
"""
wanted = normalise(triggers)
if not wanted:
msg = "a person needs at least one trigger"
raise ConfigError(msg)
taken = reserved() & {fold(trigger) for trigger in wanted}
if taken:
msg = f"already a command: {sorted(taken)}"
raise ConfigError(msg)
path = directory / PEOPLE_FILE
document = _document(path)
people = list(document.get("people", []))
key = next_key(people)
document["people"] = [*people, {"key": key, "triggers": wanted}]
return _commit(directory, path, document, key)
def merge(document: dict[str, Any], people: Sequence[dict[str, Any]]) -> list[dict[str, Any]]:
"""The roster with ``people`` folded in, existing keys left where they are.
Used by the setup form, which used to replace the file wholesale. Anyone
already present keeps their key and gains any new triggers; anyone new is
appended with a fresh key.
"""
merged = [dict(person) for person in document.get("people", [])]
by_key = {person.get("key"): person for person in merged}
for person in people:
existing = by_key.get(person.get("key"))
if existing is None:
fresh = dict(person)
fresh["key"] = next_key(merged)
merged.append(fresh)
by_key[fresh["key"]] = fresh
continue
seen = set(existing.get("triggers", []))
existing["triggers"] = [
*existing.get("triggers", []),
*(t for t in person.get("triggers", []) if t not in seen),
]
return merged
def _commit(directory: Path, path: Path, document: dict[str, Any], key: str) -> str:
"""Write the roster, or put back exactly what was there and re-raise."""
previous = path.read_bytes() if path.exists() else None
write_json(path, document)
try:
load(directory)
except ConfigError:
if previous is None:
path.unlink(missing_ok=True)
else:
path.write_bytes(previous)
raise
return key
def _document(path: Path) -> dict[str, Any]:
"""The roster as it stands.
Strict, unlike the ledger's reader: this file *is* the album's manifest,
and appending to a document that failed to parse would write a roster
that silently dropped everyone already in it.
Raises:
ConfigError: The file is not a JSON object.
"""
if not path.exists():
return {}
try:
document = json.loads(path.read_text(encoding="utf-8"))
except json.JSONDecodeError as error:
msg = f"{path.name} is not valid JSON: {error}"
raise ConfigError(msg) from error
if not isinstance(document, dict):
msg = f"{path.name} is not an object"
raise ConfigError(msg)
return document

101
signal_bot/pack_report.py Normal file
View File

@ -0,0 +1,101 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""What an import did, in buckets that have to add up.
Split from pack_import.py at the shared 250-line cap, along the seam between
deciding what happens to a file and saying what happened. The arithmetic is
the point: every source file lands in exactly one bucket, and the printed
total is computed from the decisions rather than tracked alongside them, so a
report that does not add up is not expressible.
Every bucket prints even at zero. A suppressed zero is a silently missing
category -- "0 could not be stripped" is how an operator sees that the video
pipeline is still working.
"""
from __future__ import annotations
from dataclasses import dataclass
from enum import StrEnum
from typing import TYPE_CHECKING, Final
from signal_bot.pack import MEDIA_DIR
if TYPE_CHECKING:
from pathlib import Path
# Imported for typing only: pack_import imports this module at run time,
# and a real import here would close the loop.
from signal_bot.pack_import import Decision
SKIPPED_DIR: Final = "_skipped"
class Outcome(StrEnum):
"""Why one source file did or did not end up in the album."""
NOT_MEDIA = "not-media"
UNSTRIPPABLE = "unstrippable"
ALREADY_IN_PACK = "already-in-pack"
EXACT_DUP = "exact-dup"
IMPORTED = "imported"
# Printed in this order, so the report reads as a funnel.
ORDER: Final = (
Outcome.NOT_MEDIA,
Outcome.UNSTRIPPABLE,
Outcome.ALREADY_IN_PACK,
Outcome.EXACT_DUP,
Outcome.IMPORTED,
)
LABELS: Final[dict[Outcome, str]] = {
Outcome.NOT_MEDIA: "not media",
Outcome.UNSTRIPPABLE: "could not be stripped",
Outcome.ALREADY_IN_PACK: "already in pack",
Outcome.EXACT_DUP: "duplicate in batch",
Outcome.IMPORTED: "imported",
}
@dataclass(frozen=True, slots=True)
class ImportReport:
"""Every source file, bucketed, plus what it does to the album."""
source: Path
directory: Path
decisions: tuple[Decision, ...]
before: int
@property
def counts(self) -> dict[Outcome, int]:
"""How many files fell into each bucket, every bucket present."""
tally = dict.fromkeys(ORDER, 0)
for decision in self.decisions:
tally[decision.outcome] += 1
return tally
@property
def imported(self) -> tuple[Decision, ...]:
"""The decisions that put a new file in the album."""
return tuple(d for d in self.decisions if d.outcome is Outcome.IMPORTED)
@property
def after(self) -> int:
"""How many files media/ holds once this import is applied."""
return self.before + len(self.imported)
def lines(self) -> list[str]:
"""The report, with a total that has to add up to what was seen."""
counts = self.counts
lines = [f"== {self.source} -> {self.directory}", ""]
lines.append(f"{len(self.decisions):6} files seen")
for outcome in ORDER:
note = "" if outcome is Outcome.IMPORTED else f" -> {SKIPPED_DIR}/{outcome}/"
lines.append(f"{counts[outcome]:6} {LABELS[outcome]:<21}{note}")
lines.append(" ---")
lines.append(
f"{sum(counts.values()):6} accounted for"
f" {self.before} -> {self.after} in {MEDIA_DIR}/"
)
return lines

64
tests/test_media_strip.py
View File

@ -70,8 +70,14 @@ def test_a_clean_file_comes_back_stripped(tmp_path: Path) -> None:
assert stripper.strip(b"\xff\xd8\xffdata", JPEG) == b"\xff\xd8\xffdata"
def test_a_tag_that_survived_the_strip_refuses_the_upload(tmp_path: Path) -> None:
def test_a_tag_that_survived_the_strip_refuses_the_upload(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""The whole point: the exit code was 0 and the file is still dirty."""
# flatten is stubbed out so this reaches the tag check. Without it the
# re-mux refuses the fixture first -- it is twelve bytes, not a video --
# and the test would still pass, for the wrong reason.
monkeypatch.setattr("signal_bot.media_strip.flatten", lambda payload, suffix: payload)
stripper = ExiftoolStripper(_stub(tmp_path, "leaky", strip=COPIES, report=LEAKY_REPORT))
assert stripper.strip(b"\x00\x00\x00\x20ftypmp42", MP4) is None
@ -168,3 +174,59 @@ def test_locate_returns_none_when_it_really_is_absent(
def test_the_arch_vendor_perl_path_is_on_the_fallback_list() -> None:
"""Named explicitly: this is the one that actually bit."""
assert "/usr/bin/vendor_perl/exiftool" in FALLBACK_PATHS
RECORDS_ARGV = (
'open("ARGV_PATH", "w").write(" ".join(args)); '
'open(args[args.index("-o") + 1], "wb").write(open(args[-1], "rb").read())'
)
def test_the_date_tags_are_zeroed_explicitly(tmp_path: Path) -> None:
# media_tags allows the QuickTime date tags "by value", on the stated
# grounds that -all= zeroes rather than deletes them. Measured on
# 2026-08-31 that is false: -all= alone left QuickTime:CreateDate at its
# real value and every mp4 in the live album was refused because of it.
# -time:all= is what actually zeroes them, so it must stay in the argv.
argv = tmp_path / "argv.txt"
stripper = ExiftoolStripper(
_stub(tmp_path, "records", strip=RECORDS_ARGV.replace("ARGV_PATH", str(argv)))
)
stripper.strip(b"\xff\xd8\xffdata", JPEG)
assert "-time:all=" in argv.read_text(encoding="utf-8")
def test_a_video_has_its_metadata_tracks_dropped_first(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
# exiftool cannot see a mebx track at all -- see media_tracks for the
# face-detection payload that survives a complete tag strip -- so the
# re-mux has to happen here or not at all.
seen: list[tuple[bytes, str]] = []
def fake_flatten(payload: bytes, suffix: str) -> bytes:
seen.append((payload, suffix))
return b"flattened"
monkeypatch.setattr("signal_bot.media_strip.flatten", fake_flatten)
ExiftoolStripper(_stub(tmp_path, "ok1", strip=COPIES)).strip(b"payload", MP4)
assert seen == [(b"payload", ".mp4")]
def test_a_still_image_is_not_sent_through_the_remux(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
def explode(payload: bytes, suffix: str) -> bytes:
raise AssertionError
monkeypatch.setattr("signal_bot.media_strip.flatten", explode)
stripper = ExiftoolStripper(_stub(tmp_path, "ok2", strip=COPIES))
assert stripper.strip(b"\xff\xd8\xffdata", JPEG) is not None
def test_a_video_whose_tracks_cannot_be_dropped_is_refused(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr("signal_bot.media_strip.flatten", lambda payload, suffix: None)
stripper = ExiftoolStripper(_stub(tmp_path, "ok3", strip=COPIES))
assert stripper.strip(b"payload", MP4) is None

142
tests/test_media_tracks.py Normal file
View File

@ -0,0 +1,142 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Dropping metadata tracks exiftool cannot see, and proving they are gone."""
from __future__ import annotations
from pathlib import Path
import pytest
from signal_bot import media_tracks
from signal_bot.external import Result
class FakeRun:
"""Stands in for signal_bot.external.run, one canned answer per binary."""
def __init__(self, ffmpeg: Result, ffprobe: Result | None = None) -> None:
self.ffmpeg = ffmpeg
self.ffprobe = ffprobe or Result(ok=True, stdout="video\naudio\n", stderr="")
self.commands: list[list[str]] = []
def __call__(self, command: list[str], *, timeout: float = 0) -> Result:
"""Answer as the binary the command names, and write the output file."""
self.commands.append(list(command))
if "ffprobe" in command[0]:
return self.ffprobe
if self.ffmpeg.ok:
# The real ffmpeg leaves a file behind, and the code checks for it.
Path(command[-1]).write_bytes(b"remuxed")
return self.ffmpeg
@pytest.fixture
def found(monkeypatch: pytest.MonkeyPatch) -> None:
"""Both binaries present, so tests exercise the logic and not locate()."""
monkeypatch.setattr(media_tracks, "locate", lambda binary: f"/usr/bin/{binary}")
def _use(monkeypatch: pytest.MonkeyPatch, runner: FakeRun) -> None:
monkeypatch.setattr(media_tracks, "run", runner)
def test_a_clean_remux_returns_the_new_bytes(monkeypatch: pytest.MonkeyPatch, found: None) -> None:
_use(monkeypatch, FakeRun(Result(ok=True, stdout="", stderr="")))
assert media_tracks.flatten(b"video", ".mp4") == b"remuxed"
def test_audio_is_mapped_optionally(monkeypatch: pytest.MonkeyPatch, found: None) -> None:
# "0:a?" and not "0:a": a clip with no sound is still a clip.
runner = FakeRun(Result(ok=True, stdout="", stderr=""))
_use(monkeypatch, runner)
media_tracks.flatten(b"video", ".mp4")
assert "0:a?" in runner.commands[0]
def test_a_missing_ffmpeg_refuses_rather_than_storing(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(media_tracks, "locate", lambda binary: None)
assert media_tracks.flatten(b"video", ".mp4") is None
def test_a_missing_ffprobe_refuses(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(
media_tracks, "locate", lambda binary: None if binary == "ffprobe" else "/usr/bin/ffmpeg"
)
assert media_tracks.flatten(b"video", ".mp4") is None
def test_an_ffmpeg_failure_refuses(monkeypatch: pytest.MonkeyPatch, found: None) -> None:
_use(monkeypatch, FakeRun(Result(ok=False, stdout="", stderr="boom")))
assert media_tracks.flatten(b"video", ".mp4") is None
def test_a_surviving_data_stream_refuses(monkeypatch: pytest.MonkeyPatch, found: None) -> None:
# The whole point: a metadata track that outlived the re-mux must not be
# stored, because no tag allowlist downstream can see it.
_use(
monkeypatch,
FakeRun(
Result(ok=True, stdout="", stderr=""),
Result(ok=True, stdout="video\naudio\ndata\n", stderr=""),
),
)
assert media_tracks.flatten(b"video", ".mp4") is None
def test_an_unreadable_result_refuses(monkeypatch: pytest.MonkeyPatch, found: None) -> None:
_use(
monkeypatch,
FakeRun(Result(ok=True, stdout="", stderr=""), Result(ok=False, stdout="", stderr="x")),
)
assert media_tracks.flatten(b"video", ".mp4") is None
def test_a_file_ffprobe_describes_as_nothing_refuses(
monkeypatch: pytest.MonkeyPatch, found: None
) -> None:
# Empty is a failure, not "no bad streams": nobody can vouch for it.
_use(
monkeypatch,
FakeRun(Result(ok=True, stdout="", stderr=""), Result(ok=True, stdout="\n", stderr="")),
)
assert media_tracks.flatten(b"video", ".mp4") is None
def test_streams_strips_the_trailing_comma_ffprobe_emits(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
_use(monkeypatch, FakeRun(Result(ok=True, stdout="", stderr="")))
assert media_tracks.streams(tmp_path / "x.mp4", "/usr/bin/ffprobe") == ("video", "audio")
def test_streams_reports_failure_as_none(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
_use(
monkeypatch,
FakeRun(Result(ok=True, stdout="", stderr=""), Result(ok=False, stdout="", stderr="")),
)
assert media_tracks.streams(tmp_path / "x.mp4", "/usr/bin/ffprobe") is None
def test_locate_prefers_path(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(media_tracks.shutil, "which", lambda binary: f"/somewhere/{binary}")
assert media_tracks.locate("ffmpeg") == "/somewhere/ffmpeg"
def test_locate_falls_back_to_the_known_directories(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
# A systemd unit's PATH is "/usr/local/bin:/usr/bin" and nothing else.
monkeypatch.setattr(media_tracks.shutil, "which", lambda binary: None)
(tmp_path / "ffmpeg").write_text("#!/bin/sh\n", encoding="utf-8")
monkeypatch.setattr(media_tracks, "FALLBACK_DIRS", (str(tmp_path),))
assert media_tracks.locate("ffmpeg") == str(tmp_path / "ffmpeg")
def test_locate_reports_nothing_when_it_is_not_installed(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
monkeypatch.setattr(media_tracks.shutil, "which", lambda binary: None)
monkeypatch.setattr(media_tracks, "FALLBACK_DIRS", (str(tmp_path),))
assert media_tracks.locate("ffmpeg") is None

107
tests/test_pack_album.py Normal file
View File

@ -0,0 +1,107 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""The album as the tagging page sees it."""
from __future__ import annotations
import json
from typing import TYPE_CHECKING
import pytest
from signal_bot.pack import MEDIA_DIR, PEOPLE_FILE, TAGS_FILE
from signal_bot.pack_album import Album
from tests.conftest import WATCHED_GROUP
from tests.pack_fixture import CLIP, PIXEL
if TYPE_CHECKING:
from pathlib import Path
def _album(tmp_path: Path, *, untagged_only: bool = False) -> Album:
media = tmp_path / MEDIA_DIR
media.mkdir(parents=True)
(media / "one.png").write_bytes(PIXEL)
(media / "two.png").write_bytes(PIXEL)
(media / "clip.mp4").write_bytes(CLIP)
(tmp_path / PEOPLE_FILE).write_text(
json.dumps({"groups": [WATCHED_GROUP], "people": [{"key": "p1", "triggers": ["!alfa"]}]}),
encoding="utf-8",
)
(tmp_path / TAGS_FILE).write_text(json.dumps({"one.png": ["p1"]}), encoding="utf-8")
return Album(tmp_path, untagged_only=untagged_only)
def test_every_taggable_file_is_listed_in_a_stable_order(tmp_path: Path) -> None:
assert _album(tmp_path).files == ["clip.mp4", "one.png", "two.png"]
def test_the_state_carries_the_filter_as_a_flag_not_a_shorter_list(tmp_path: Path) -> None:
# One list, one index space. /media/{index} and /api/tag both index
# files, so filtering server-side would show item i of the album while
# tagging item i of the filtered set -- every photo tagged with the
# wrong people, and no error anywhere.
state = _album(tmp_path, untagged_only=True).state()
assert state["untagged_only"] is True
assert [entry["name"] for entry in state["files"]] == ["clip.mp4", "one.png", "two.png"]
assert [entry["keys"] for entry in state["files"]] == [[], ["p1"], []]
def test_an_album_with_no_roster_is_not_configured(tmp_path: Path) -> None:
(tmp_path / MEDIA_DIR).mkdir(parents=True)
assert Album(tmp_path).state()["configured"] is False
def test_tagging_records_who_is_in_a_file(tmp_path: Path) -> None:
album = _album(tmp_path)
album.tag("two.png", ["p1"])
assert album.read(TAGS_FILE)["two.png"] == ["p1"]
def test_tagging_a_file_the_album_does_not_have_is_refused(tmp_path: Path) -> None:
with pytest.raises(KeyError):
_album(tmp_path).tag("../etc/passwd", ["p1"])
def test_a_manifest_that_is_not_there_reads_as_empty(tmp_path: Path) -> None:
(tmp_path / MEDIA_DIR).mkdir(parents=True)
assert Album(tmp_path).read("nothing.json") == {}
def test_the_summary_counts_what_was_never_opened(tmp_path: Path) -> None:
lines = _album(tmp_path).summary()
assert "3 items, 2 never opened" in lines[0]
def test_the_summary_reports_a_broken_pack_rather_than_raising(tmp_path: Path) -> None:
# A half-built pack is the normal state to stop the tagger in.
(tmp_path / MEDIA_DIR).mkdir(parents=True)
(tmp_path / PEOPLE_FILE).write_text(json.dumps({"groups": []}), encoding="utf-8")
assert any("does not load yet" in line for line in Album(tmp_path).summary())
def test_the_summary_marks_a_command_with_no_photos(tmp_path: Path) -> None:
# A command with zero is dead on arrival, and this is where it shows up
# rather than in the group.
album = _album(tmp_path)
(tmp_path / TAGS_FILE).write_text(json.dumps({}), encoding="utf-8")
assert any(line.startswith("!!") and "!alfa" in line for line in album.summary())
def test_a_person_can_be_added_to_a_configured_album(tmp_path: Path) -> None:
album = _album(tmp_path)
assert album.add_person(["delta"]) == "p2"
assert album.state()["people"][1]["triggers"] == ["!delta"]
def test_setup_merges_rather_than_replacing(tmp_path: Path) -> None:
album = _album(tmp_path)
album.setup([WATCHED_GROUP], [{"triggers": ["!delta"]}], [])
keys = [person["key"] for person in album.state()["people"]]
assert keys == ["p1", "p2"]
def test_a_write_is_atomic_and_leaves_no_temporary(tmp_path: Path) -> None:
album = _album(tmp_path)
album.write("thing.json", {"a": 1})
assert json.loads((tmp_path / "thing.json").read_text(encoding="utf-8")) == {"a": 1}
assert not (tmp_path / "thing.tmp").exists()

238
tests/test_pack_import.py Normal file
View File

@ -0,0 +1,238 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Importing a folder: the two-stage dedup, and the arithmetic that proves it."""
from __future__ import annotations
from typing import TYPE_CHECKING
from signal_bot import pack_ledger
from signal_bot.media_sniff import MediaKind
from signal_bot.pack import MEDIA_DIR
from signal_bot.pack_custom import FILE_MODE
from signal_bot.pack_import import plan_import, run_import
from signal_bot.pack_report import SKIPPED_DIR, Outcome
from tests.pack_fixture import CLIP, PIXEL
if TYPE_CHECKING:
from pathlib import Path
# A second distinct PNG, so a test can have two files that are genuinely
# different rather than two names for one.
OTHER = PIXEL.replace(b"IDAT", b"IDAT", 1) + b"\x00"
class FakeStripper:
"""Strips by mapping bytes, so no test ever shells out to exiftool."""
def __init__(self, mapping: dict[bytes, bytes | None] | None = None) -> None:
self.mapping = mapping or {}
def strip(self, payload: bytes, kind: MediaKind) -> bytes | None:
"""The mapped output, or the payload unchanged."""
assert isinstance(kind, MediaKind)
return self.mapping.get(payload, payload)
def _album(tmp_path: Path, files: dict[str, bytes] | None = None) -> Path:
directory = tmp_path / "pack"
media = directory / MEDIA_DIR
media.mkdir(parents=True)
for name, payload in (files or {}).items():
(media / name).write_bytes(payload)
return directory
def _source(tmp_path: Path, files: dict[str, bytes]) -> Path:
source = tmp_path / "batch" / "Folder"
source.mkdir(parents=True)
for name, payload in files.items():
(source / name).write_bytes(payload)
return source
def test_every_file_lands_in_exactly_one_bucket(tmp_path: Path) -> None:
# The done condition, mechanised: the buckets must sum to what was seen.
directory = _album(tmp_path, {"have.png": PIXEL})
source = _source(
tmp_path,
{"new.png": OTHER, "same.png": PIXEL, "copy.png": OTHER, "notes.txt": b"hello"},
)
report = plan_import(source, directory, FakeStripper())
assert sum(report.counts.values()) == len(report.decisions) == 4
assert report.counts == {
Outcome.NOT_MEDIA: 1,
Outcome.UNSTRIPPABLE: 0,
Outcome.ALREADY_IN_PACK: 1,
Outcome.EXACT_DUP: 1,
Outcome.IMPORTED: 1,
}
assert report.before == 1
assert report.after == 2
def test_the_report_prints_every_bucket_including_the_empty_ones(tmp_path: Path) -> None:
# A suppressed zero is a silently missing category; "0 could not be
# stripped" is how the operator sees the video fix still holding.
report = plan_import(_source(tmp_path, {}), _album(tmp_path), FakeStripper())
body = "\n".join(report.lines())
for outcome in Outcome:
assert str(outcome) in body or outcome is Outcome.IMPORTED
assert "could not be stripped" in body
assert "accounted for" in body
def test_a_file_already_in_the_album_is_not_imported_twice(tmp_path: Path) -> None:
directory = _album(tmp_path, {"have.png": PIXEL})
source = _source(tmp_path, {"renamed.png": PIXEL})
report = plan_import(source, directory, FakeStripper())
assert report.decisions[0].outcome is Outcome.ALREADY_IN_PACK
def test_two_sources_that_strip_to_one_photo_collapse(tmp_path: Path) -> None:
# The measured case that a source-hash-only ledger would import twice:
# the same photo re-encoded, identical once the metadata is gone.
directory = _album(tmp_path)
source = _source(tmp_path, {"a.png": PIXEL, "b.png": OTHER})
stripper = FakeStripper({PIXEL: PIXEL, OTHER: PIXEL})
report = plan_import(source, directory, stripper)
outcomes = [d.outcome for d in report.decisions]
assert outcomes == [Outcome.IMPORTED, Outcome.EXACT_DUP]
def test_a_chat_upload_and_an_exported_copy_collapse(tmp_path: Path) -> None:
# Same as above but across runs: the album already holds the stripped
# bytes, and the export folder holds the unstripped original.
directory = _album(tmp_path, {"hash.png": PIXEL})
source = _source(tmp_path, {"original.png": OTHER})
report = plan_import(source, directory, FakeStripper({OTHER: PIXEL}))
assert report.decisions[0].outcome is Outcome.ALREADY_IN_PACK
def test_a_file_the_stripper_will_not_vouch_for_is_refused(tmp_path: Path) -> None:
report = plan_import(
_source(tmp_path, {"a.png": PIXEL}), _album(tmp_path), FakeStripper({PIXEL: None})
)
assert report.decisions[0].outcome is Outcome.UNSTRIPPABLE
def test_format_is_decided_by_bytes_not_by_suffix(tmp_path: Path) -> None:
directory = _album(tmp_path)
source = _source(tmp_path, {"photo.txt": PIXEL, "fake.jpg": b"not an image at all"})
report = plan_import(source, directory, FakeStripper())
by_name = {d.source.name: d for d in report.decisions}
assert by_name["fake.jpg"].outcome is Outcome.NOT_MEDIA
imported = by_name["photo.txt"]
assert imported.outcome is Outcome.IMPORTED
# The suffix is corrected to match the bytes: pack.media_type looks the
# stored suffix up in a closed table, and a miss fails the *next* load.
assert imported.stored is not None
assert imported.stored.name == "photo.png"
def test_a_suffix_that_agrees_with_the_bytes_is_kept(tmp_path: Path) -> None:
# .jpeg and .jpg are both image/jpeg, so a .jpeg keeps its name.
jpeg = b"\xff\xd8\xff" + b"\x00" * 40
source = _source(tmp_path, {"photo.jpeg": jpeg})
report = plan_import(source, _album(tmp_path), FakeStripper())
assert report.decisions[0].stored is not None
assert report.decisions[0].stored.name == "photo.jpeg"
def test_a_name_collision_disambiguates_rather_than_dropping_a_photo(tmp_path: Path) -> None:
# The old bug: two exports both saying IMG_0001.jpg silently lost one.
directory = _album(tmp_path, {"IMG_1.png": PIXEL})
source = _source(tmp_path, {"IMG_1.png": OTHER})
report = plan_import(source, directory, FakeStripper())
assert report.decisions[0].stored is not None
assert report.decisions[0].stored.name == "IMG_1-2.png"
def test_video_is_imported_like_anything_else(tmp_path: Path) -> None:
report = plan_import(_source(tmp_path, {"c.mp4": CLIP}), _album(tmp_path), FakeStripper())
assert report.decisions[0].outcome is Outcome.IMPORTED
def test_a_subdirectory_in_the_source_is_ignored(tmp_path: Path) -> None:
source = _source(tmp_path, {"a.png": PIXEL})
(source / "nested").mkdir()
report = plan_import(source, _album(tmp_path), FakeStripper())
assert len(report.decisions) == 1
def test_running_the_import_stores_the_new_file_and_the_ledger(tmp_path: Path) -> None:
directory = _album(tmp_path)
source = _source(tmp_path, {"a.png": PIXEL})
run_import(source, directory, FakeStripper())
stored = directory / MEDIA_DIR / "a.png"
assert stored.read_bytes() == PIXEL
assert stored.stat().st_mode & 0o777 == FILE_MODE
assert pack_ledger.read(directory)["a.png"].source == pack_ledger.digest(PIXEL)
def test_a_rejected_file_is_moved_and_never_deleted(tmp_path: Path) -> None:
directory = _album(tmp_path, {"have.png": PIXEL})
source = _source(tmp_path, {"same.png": PIXEL, "notes.txt": b"hello"})
run_import(source, directory, FakeStripper())
parent = source.parent
assert (parent / SKIPPED_DIR / "already-in-pack" / "same.png").read_bytes() == PIXEL
assert (parent / SKIPPED_DIR / "not-media" / "notes.txt").read_bytes() == b"hello"
assert not (source / "same.png").exists()
def test_the_skipped_folder_sits_beside_the_source_not_under_home(tmp_path: Path) -> None:
# Derived from the source argument so a tmp_path source keeps every write
# inside tmp_path -- conftest redirects XDG_DATA_HOME, not $HOME.
source = _source(tmp_path, {"notes.txt": b"hello"})
run_import(source, _album(tmp_path), FakeStripper())
assert (source.parent / SKIPPED_DIR).is_dir()
def test_importing_the_same_folder_twice_is_a_no_op(tmp_path: Path) -> None:
# The property the ledger exists for.
directory = _album(tmp_path)
source = _source(tmp_path, {"a.png": PIXEL})
run_import(source, directory, FakeStripper())
again = _source(tmp_path / "second", {"a.png": PIXEL})
report = run_import(again, directory, FakeStripper())
assert report.counts[Outcome.IMPORTED] == 0
assert report.counts[Outcome.ALREADY_IN_PACK] == 1
def test_an_import_creates_the_album_directory_when_there_is_none(tmp_path: Path) -> None:
directory = tmp_path / "fresh"
directory.mkdir()
report = run_import(_source(tmp_path, {"a.png": PIXEL}), directory, FakeStripper())
assert (directory / MEDIA_DIR / "a.png").exists()
assert report.before == 0
def test_the_import_does_not_write_tags(tmp_path: Path) -> None:
# An imported file is simply absent from tags.json, which is what
# --untagged keys off and what pack_load already tolerates.
directory = _album(tmp_path)
run_import(_source(tmp_path, {"a.png": PIXEL}), directory, FakeStripper())
assert not (directory / "tags.json").exists()
def test_a_photo_that_arrives_mid_run_is_not_stored_twice(tmp_path: Path) -> None:
# Planning strips every candidate, which takes minutes on a real batch.
# A photo uploaded from the chat during that window is in the album by
# the time the apply pass runs, and storing it anyway puts it there
# twice: once hash-named by !add_photo, once under its original name.
# Measured: that happened to 6 photos during the first live import.
directory = _album(tmp_path)
source = _source(tmp_path, {"a.png": PIXEL})
class Interloper(FakeStripper):
"""Drops the photo into the album the moment planning reads it."""
def strip(self, payload: bytes, kind: MediaKind) -> bytes | None:
(directory / MEDIA_DIR / "from-the-chat.png").write_bytes(payload)
return super().strip(payload, kind)
report = run_import(source, directory, Interloper())
assert report.counts[Outcome.IMPORTED] == 0
assert report.counts[Outcome.ALREADY_IN_PACK] == 1
assert sorted(p.name for p in (directory / MEDIA_DIR).iterdir()) == ["from-the-chat.png"]
# And it is moved aside like any other repeat, never left behind.
assert (source.parent / SKIPPED_DIR / "already-in-pack" / "a.png").exists()

133
tests/test_pack_ledger.py Normal file
View File

@ -0,0 +1,133 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""The import ledger: what the album has already taken in, and from what."""
from __future__ import annotations
import json
from typing import TYPE_CHECKING
from signal_bot import pack_ledger
from signal_bot.pack import MEDIA_DIR
from signal_bot.pack_custom import FILE_MODE
from signal_bot.pack_ledger import LEDGER_FILE, Entry
from signal_bot.pack_store import ADDS_FILE
from tests.pack_fixture import PIXEL
if TYPE_CHECKING:
from pathlib import Path
def _album(tmp_path: Path, files: dict[str, bytes]) -> Path:
"""A pack directory holding exactly ``files`` in media/."""
media = tmp_path / MEDIA_DIR
media.mkdir(parents=True)
for name, payload in files.items():
(media / name).write_bytes(payload)
return tmp_path
def test_digest_is_a_full_sha256() -> None:
# Not content_name's truncated 32: a filename is a name, this is identity.
assert len(pack_ledger.digest(b"x")) == 64
def test_a_missing_ledger_reads_as_empty(tmp_path: Path) -> None:
assert pack_ledger.read(tmp_path) == {}
def test_a_corrupt_ledger_degrades_instead_of_raising(tmp_path: Path) -> None:
# The whole point: pack_load raises at startup and Restart=always turns
# that into a loop. A broken ledger must cost a duplicate, not the bot.
(tmp_path / LEDGER_FILE).write_text("{not json", encoding="utf-8")
assert pack_ledger.read(tmp_path) == {}
def test_a_ledger_that_is_not_an_object_reads_as_empty(tmp_path: Path) -> None:
(tmp_path / LEDGER_FILE).write_text("[1, 2]", encoding="utf-8")
assert pack_ledger.read(tmp_path) == {}
def test_one_malformed_row_does_not_lose_the_others(tmp_path: Path) -> None:
(tmp_path / LEDGER_FILE).write_text(
json.dumps(
{
"good.jpg": {"source": "a" * 64, "stored": "b" * 64},
"not-a-dict.jpg": "nonsense",
"no-stored.jpg": {"source": "c" * 64},
"bad-source.jpg": {"source": 7, "stored": "d" * 64},
}
),
encoding="utf-8",
)
assert set(pack_ledger.read(tmp_path)) == {"good.jpg"}
def test_a_null_source_round_trips(tmp_path: Path) -> None:
entries = {"chat.jpg": Entry(source=None, stored="e" * 64)}
pack_ledger.write(tmp_path, entries)
assert pack_ledger.read(tmp_path) == entries
def test_the_ledger_is_written_private(tmp_path: Path) -> None:
pack_ledger.write(tmp_path, {"a.jpg": Entry(source="1" * 64, stored="2" * 64)})
assert (tmp_path / LEDGER_FILE).stat().st_mode & 0o777 == FILE_MODE
def test_source_hashes_skips_the_unrecoverable_ones() -> None:
entries = {
"a.jpg": Entry(source="1" * 64, stored="2" * 64),
"b.jpg": Entry(source=None, stored="3" * 64),
}
assert pack_ledger.source_hashes(entries) == {"1" * 64}
assert pack_ledger.stored_hashes(entries) == {"2" * 64, "3" * 64}
def test_backfill_records_an_imported_file_as_its_own_source(tmp_path: Path) -> None:
# shutil.copy2 copied these verbatim, so the bytes on disk *are* the
# source bytes. This is the only moment that is recoverable.
directory = _album(tmp_path, {"IMG_1.jpg": PIXEL})
entries = pack_ledger.backfill(directory)
assert entries["IMG_1.jpg"].source == entries["IMG_1.jpg"].stored
assert entries["IMG_1.jpg"].stored == pack_ledger.digest(PIXEL)
def test_backfill_leaves_a_chat_upload_without_a_source(tmp_path: Path) -> None:
# adds.json is the exact record of what came in through !add_photo, and
# those were stripped on the way in: the source bytes never existed here.
directory = _album(tmp_path, {"abc123.jpg": PIXEL})
(directory / ADDS_FILE).write_text(
json.dumps({"abc123.jpg": {"by": "someone", "at": 1}}), encoding="utf-8"
)
entries = pack_ledger.backfill(directory)
assert entries["abc123.jpg"].source is None
assert entries["abc123.jpg"].stored == pack_ledger.digest(PIXEL)
def test_backfill_persists_what_it_found(tmp_path: Path) -> None:
directory = _album(tmp_path, {"IMG_1.jpg": PIXEL})
pack_ledger.backfill(directory)
assert set(pack_ledger.read(directory)) == {"IMG_1.jpg"}
def test_backfill_never_revises_an_entry_it_already_has(tmp_path: Path) -> None:
# Re-running must not overwrite a source hash recorded before a strip
# with the post-strip bytes, which would silently lose the origin.
directory = _album(tmp_path, {"IMG_1.jpg": PIXEL})
pack_ledger.write(directory, {"IMG_1.jpg": Entry(source="old", stored="older")})
entries = pack_ledger.backfill(directory)
assert entries["IMG_1.jpg"] == Entry(source="old", stored="older")
def test_backfill_is_idempotent(tmp_path: Path) -> None:
directory = _album(tmp_path, {"IMG_1.jpg": PIXEL})
assert pack_ledger.backfill(directory) == pack_ledger.backfill(directory)
def test_backfill_ignores_a_directory_inside_media(tmp_path: Path) -> None:
directory = _album(tmp_path, {"IMG_1.jpg": PIXEL})
(directory / MEDIA_DIR / "subdir").mkdir()
assert set(pack_ledger.backfill(directory)) == {"IMG_1.jpg"}
def test_backfill_on_an_album_with_no_media_directory(tmp_path: Path) -> None:
assert pack_ledger.backfill(tmp_path) == {}

140
tests/test_pack_people.py Normal file
View File

@ -0,0 +1,140 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Appending to a roster that already exists, without renumbering it."""
from __future__ import annotations
import json
from typing import TYPE_CHECKING, Any
import pytest
from signal_bot.groups import ConfigError
from signal_bot.pack import PEOPLE_FILE
from signal_bot.pack_custom import CUSTOM_FILE
from signal_bot.pack_people import add_person, merge, next_key, normalise, reserved
from tests.conftest import WATCHED_GROUP
if TYPE_CHECKING:
from pathlib import Path
def _roster(tmp_path: Path, people: list[dict[str, Any]]) -> Path:
(tmp_path / PEOPLE_FILE).write_text(
json.dumps({"groups": [WATCHED_GROUP], "people": people}), encoding="utf-8"
)
return tmp_path
def test_the_next_key_is_one_past_the_highest_not_the_count() -> None:
# p2 deleted by hand: a count-based key would hand out p3 again and the
# new person would silently inherit the old p3's photos.
assert next_key([{"key": "p1"}, {"key": "p3"}]) == "p4"
def test_the_next_key_on_an_empty_roster() -> None:
assert next_key([]) == "p1"
def test_the_next_key_ignores_entries_that_are_not_keys() -> None:
assert next_key([{"key": "pX"}, {"key": 7}, {}, {"key": "p2"}]) == "p3"
def test_a_trigger_may_be_typed_with_or_without_the_bang() -> None:
assert normalise([" alfa ", "!beta", "", " "]) == ["!alfa", "!beta"]
def test_the_bot_s_own_commands_are_known() -> None:
# Folded, and folding keeps the "!" -- which is why add_person compares
# against normalise()'s "!"-prefixed spelling and not the bare word.
assert "!kot" in reserved()
assert not any(word.startswith("_") for word in reserved())
def test_a_person_is_appended_with_a_stable_key(tmp_path: Path) -> None:
directory = _roster(tmp_path, [{"key": "p1", "triggers": ["!alfa"]}])
assert add_person(directory, ["delta"]) == "p2"
document = json.loads((directory / PEOPLE_FILE).read_text(encoding="utf-8"))
assert document["people"][0] == {"key": "p1", "triggers": ["!alfa"]}
assert document["people"][1] == {"key": "p2", "triggers": ["!delta"]}
def test_a_person_can_be_added_to_an_album_with_no_roster_yet(tmp_path: Path) -> None:
(tmp_path / PEOPLE_FILE).write_text(json.dumps({"groups": [WATCHED_GROUP]}), encoding="utf-8")
assert add_person(tmp_path, ["!alfa"]) == "p1"
def test_a_person_with_no_triggers_is_refused(tmp_path: Path) -> None:
directory = _roster(tmp_path, [{"key": "p1", "triggers": ["!alfa"]}])
with pytest.raises(ConfigError, match="at least one trigger"):
add_person(directory, [" "])
def test_a_trigger_that_is_already_a_command_is_refused(tmp_path: Path) -> None:
# !kot would make personal_data_sources find "kot" in aliases.json and
# block every commit forever, fixable only by editing a private album.
directory = _roster(tmp_path, [{"key": "p1", "triggers": ["!alfa"]}])
with pytest.raises(ConfigError, match="already a command"):
add_person(directory, ["kot"])
def test_a_trigger_with_a_digit_is_refused_and_nothing_is_written(tmp_path: Path) -> None:
# Folding maps 0134578 onto letters, so a digit is ambiguous by design.
directory = _roster(tmp_path, [{"key": "p1", "triggers": ["!alfa"]}])
before = (directory / PEOPLE_FILE).read_bytes()
with pytest.raises(ConfigError):
add_person(directory, ["!alfa2"])
assert (directory / PEOPLE_FILE).read_bytes() == before
def test_a_trigger_colliding_with_an_existing_person_is_refused(tmp_path: Path) -> None:
directory = _roster(tmp_path, [{"key": "p1", "triggers": ["!alfa"]}])
before = (directory / PEOPLE_FILE).read_bytes()
with pytest.raises(ConfigError):
add_person(directory, ["!alfa"])
assert (directory / PEOPLE_FILE).read_bytes() == before
def test_a_trigger_colliding_with_a_custom_label_is_refused(tmp_path: Path) -> None:
directory = _roster(tmp_path, [{"key": "p1", "triggers": ["!alfa"]}])
(directory / CUSTOM_FILE).write_text(json.dumps({"t1": ["wakacje"]}), encoding="utf-8")
with pytest.raises(ConfigError):
add_person(directory, ["wakacje"])
def test_a_failed_add_restores_a_roster_that_did_not_exist(tmp_path: Path) -> None:
with pytest.raises(ConfigError):
add_person(tmp_path, ["!alfa"])
assert not (tmp_path / PEOPLE_FILE).exists()
def test_a_roster_that_is_not_json_is_refused_rather_than_overwritten(tmp_path: Path) -> None:
# Strict, unlike the ledger: appending to a document that failed to parse
# would write a roster that silently dropped everyone already in it.
(tmp_path / PEOPLE_FILE).write_text("{oops", encoding="utf-8")
with pytest.raises(ConfigError, match="not valid JSON"):
add_person(tmp_path, ["!alfa"])
def test_a_roster_that_is_not_an_object_is_refused(tmp_path: Path) -> None:
(tmp_path / PEOPLE_FILE).write_text("[]", encoding="utf-8")
with pytest.raises(ConfigError, match="not an object"):
add_person(tmp_path, ["!alfa"])
def test_merge_keeps_existing_keys_where_they_are() -> None:
# The parseSetup bug, pinned: positional keys renumber everyone after an
# insertion and repoint every tags.json entry at the wrong person.
document = {"people": [{"key": "p1", "triggers": ["!alfa"]}, {"key": "p2", "triggers": ["!b"]}]}
merged = merge(document, [{"key": "p2", "triggers": ["!b", "!beta"]}])
assert [person["key"] for person in merged] == ["p1", "p2"]
assert merged[1]["triggers"] == ["!b", "!beta"]
def test_merge_appends_someone_new_with_a_fresh_key() -> None:
document = {"people": [{"key": "p1", "triggers": ["!alfa"]}]}
merged = merge(document, [{"triggers": ["!delta"]}])
assert merged[1]["key"] == "p2"
def test_merge_on_an_empty_document() -> None:
assert merge({}, [{"triggers": ["!alfa"]}])[0]["key"] == "p1"

66
tests/test_tool_imports.py Normal file
View File

@ -0,0 +1,66 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Every script under tool/ imports, and the tagger answers --help.
This exists because ``tool/tagger.py`` shipped an unconditional ImportError
and survived two releases with it: commit 48957a1 renamed
``pack.default_dir`` to ``default_dirs`` and missed the caller. Nothing could
have caught it. ``tool/`` is outside ``--cov`` in pyproject.toml, mypy ran
with ``args: [signal_bot, fake_signal]``, and ruff lints these files but
cannot resolve a name in another module.
So this is deliberately the cheapest possible test: import each file and, for
the one with a CLI, run it. It adds no coverage burden -- ``tests/`` is not
measured -- and no module-scope ImportError can survive it.
"""
from __future__ import annotations
import importlib.util
import sys
from pathlib import Path
import pytest
from signal_bot.external import run
TOOL = Path(__file__).resolve().parent.parent / "tool"
def _modules() -> list[Path]:
"""Every Python file under tool/, so a new one is covered automatically."""
return sorted(path for path in TOOL.glob("*.py") if path.name != "__init__.py")
def test_there_is_something_to_check() -> None:
# Guards against the glob silently matching nothing and the whole file
# passing vacuously.
assert _modules()
@pytest.mark.parametrize("path", _modules(), ids=lambda path: path.name)
def test_the_module_imports(path: Path) -> None:
spec = importlib.util.spec_from_file_location(f"_tool_{path.stem}", path)
assert spec is not None
assert spec.loader is not None
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module
# tool/ is a directory of scripts, not a package, and a couple of them
# import a sibling by bare name. That resolves in real use -- running
# "python tool/x.py" puts tool/ at sys.path[0] -- so the harness has to
# reproduce it rather than the scripts having to work around it.
sys.path.insert(0, str(TOOL))
try:
spec.loader.exec_module(module)
finally:
sys.path.remove(str(TOOL))
sys.modules.pop(spec.name, None)
def test_the_tagger_answers_help() -> None:
# Through signal_bot.external, like every other subprocess here: S603 is
# suppressed for exactly one file and widening it is what that entry
# exists to prevent.
result = run([sys.executable, str(TOOL / "tagger.py"), "--help"])
assert result.ok, result.stderr
for flag in ("--import", "--untagged", "--dry-run"):
assert flag in result.stdout

100
tool/tagger.html
View File

@ -29,6 +29,9 @@
border: 1px solid #33363d; border-radius: 8px; padding: 12px;
font-family: ui-monospace, monospace; }
.hint { color: var(--dim); font-size: 14px; }
.addperson { display: flex; gap: 8px; align-items: center; margin: 4px 0 16px; }
.addperson input { background: #10131a; color: inherit; border: 1px solid #33363d;
border-radius: 8px; padding: 8px 12px; font: inherit; }
.err { color: #f7768e; }
label.group { display: block; padding: 4px 0; }
pre.sample { background: #101214; border: 1px solid #33363d; border-radius: 8px;
@ -71,6 +74,11 @@ ktokolwiek = ala | basia <span class="dim">← losowo jedna z nich</span></pr
</div>
<div class="stage" id="stage"></div>
<div class="people" id="people"></div>
<div class="addperson">
<input id="newperson" placeholder="nowa osoba: imie, ksywka" autocomplete="off">
<button id="addperson">dodaj osobę</button>
<span id="addederror" class="hint"></span>
</div>
<div class="bar">
<button id="prev">← wstecz</button>
<span id="saved" class="hint"></span>
@ -98,7 +106,12 @@ function parseSetup(text) {
.map((t) => "!" + t.trim().replace(/^!/, ""))
.filter((t) => t.length > 1);
if (!triggers.length) { continue; }
const k = `p${people.length + 1}`;
// Reuse the key this person already has. Handing out p{length+1} on
// every save renumbers everyone after an insertion, and every tags.json
// entry then points at the wrong person, silently.
const existing = (state.people || [])
.find((q) => q.triggers.some((t) => triggers.includes(t)));
const k = existing ? existing.key : `p${nextKey(people)}`;
people.push({key: k, triggers});
// Every alias, not just the first: a combination may name a person by
// whichever spelling its author remembers.
@ -124,6 +137,16 @@ function parseSetup(text) {
return {people, combos};
}
// One past the highest number in use, never the count: deleting p3 by hand
// must not let a later add reuse the key and inherit its photos. Mirrors
// pack_people.next_key on the server, which is the authority.
function nextKey(pending) {
const used = [...(state.people || []), ...pending]
.map((p) => Number(String(p.key || "").replace(/^p/, "")))
.filter((n) => Number.isInteger(n) && n > 0);
return Math.max(0, ...used) + 1;
}
function drawGroups(known, chosen) {
const box = document.getElementById("groups");
if (!known.length) {
@ -149,17 +172,45 @@ function entries() {
return [...state.people];
}
// One expression, used by both the chip label and the key handler. Those
// used to be two independent expressions -- "i === 9 ? 0 : i + 1" here and
// "key === '0' ? 9 : Number(key) - 1" in the handler -- encoding one mapping
// twice. They agreed, and would not have survived an edit.
const HOTKEYS = [..."1234567890", ..."qwertyuiop"];
// .mp4 alone left .mov/.m4v/.webm rendering as a broken <img>.
const VIDEO = [".mp4", ".mov", ".m4v", ".webm"];
function isVideo(name) {
const lower = name.toLowerCase();
return VIDEO.some((suffix) => lower.endsWith(suffix));
}
// Which items this session presents. The server never filters: /media/{index}
// and /api/tag both index the full list, so a shorter list here would show
// item i of the album while tagging item i of the filtered set.
function shown(index) {
return !state.untagged_only || !state.files[index].keys.length;
}
function step(from, direction) {
for (let i = from + direction; i >= 0 && i < state.files.length; i += direction) {
if (shown(i)) { return i; }
}
return -1;
}
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")
stage.innerHTML = isVideo(item.name)
? `<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 === 9 ? 0 : i + 1}. ${p.triggers[0].slice(1)}
${HOTKEYS[i] ? HOTKEYS[i] + ". " : ""}${p.triggers[0].slice(1)}
</div>`).join("");
for (const node of document.querySelectorAll(".person")) {
node.onclick = () => toggle(node.dataset.key);
@ -175,13 +226,15 @@ async function toggle(k) {
setTimeout(() => { document.getElementById("saved").textContent = ""; }, 800);
}
async function move(step) {
if (step > 0 && at === state.files.length - 1) {
async function move(direction) {
const next = step(at, direction);
if (next < 0) {
if (direction < 0) { return; }
// The end of the album is a screen, not a button that stops working.
await finish();
return;
}
at = Math.min(state.files.length - 1, Math.max(0, at + step));
at = next;
draw();
}
@ -196,10 +249,9 @@ 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 (/^[0-9]$/.test(event.key)) {
// 0 is the tenth, so a pack of ten is still entirely keyboard-driven.
const index = event.key === "0" ? 9 : Number(event.key) - 1;
const person = entries()[index];
else if (event.target.tagName !== "INPUT") {
const index = HOTKEYS.indexOf(event.key);
const person = index < 0 ? null : entries()[index];
if (person) { toggle(person.key); }
}
});
@ -209,6 +261,11 @@ async function start() {
if (state.configured) {
document.getElementById("tagger").hidden = false;
at = Math.max(0, state.files.findIndex((f) => !f.keys.length));
if (!shown(at)) {
const first = step(-1, 1);
if (first < 0) { await finish(); return; }
at = first;
}
draw();
return;
}
@ -227,6 +284,29 @@ async function start() {
};
}
async function addPerson() {
const field = document.getElementById("newperson");
const error = document.getElementById("addederror");
error.textContent = "";
const triggers = field.value.split(",").map((t) => t.trim()).filter(Boolean);
if (!triggers.length) { return; }
try {
await api("/api/person", {triggers});
// Re-read rather than patch in place: the server assigns the key, and
// guessing it here is how the two ends drift apart.
const fresh = await api("/api/state");
state.people = fresh.people;
field.value = "";
draw();
} catch (err) {
error.textContent = err.message;
}
}
document.getElementById("addperson").onclick = addPerson;
document.getElementById("newperson").addEventListener("keydown", (event) => {
if (event.key === "Enter") { event.stopPropagation(); addPerson(); }
});
document.getElementById("prev").onclick = () => move(-1);
document.getElementById("next").onclick = () => move(1);
document.getElementById("again").onclick = () => {

172
tool/tagger.py
View File

@ -12,7 +12,9 @@ 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
python tool/tagger.py --import ~/photos # dedup, strip and import, then tag
python tool/tagger.py --import ~/photos --dry-run # report only
python tool/tagger.py --untagged # only items with no tags yet
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
@ -22,12 +24,10 @@ 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
from typing import TYPE_CHECKING, Any, Final
import httpx
import uvicorn
@ -36,89 +36,18 @@ 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
from signal_bot.media_strip import ExiftoolStripper
from signal_bot.pack import MEDIA_DIR, default_dirs
from signal_bot.pack_album import Album
from signal_bot.pack_import import plan_import, run_import
if TYPE_CHECKING:
from signal_bot.pack_report import ImportReport
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:
# Distinct files, not the sum per option: a photo showing two of
# the three people an "either of them" entry covers is one photo
# the command can return, and summing would claim it twice.
reachable = {name for option in entry.options for name in pack.matching(option)}
mark = " " if reachable else "!!"
lines.append(f"{mark} {entry.triggers[0]}: {len(reachable)} 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:
@ -137,17 +66,13 @@ def live_groups() -> list[dict[str, str]]:
]
def build_app(album: Album) -> FastAPI:
"""The one-page application, bound to one album."""
app = FastAPI()
def roster_routes(app: FastAPI, album: Album) -> None:
"""The two endpoints that write people.json.
@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()}
Split from build_app to keep it under the complexity gate, along the
seam that matters: everything here can refuse, and refusing has to reach
the page rather than the next restart.
"""
@app.post("/api/setup")
async def setup(payload: dict[str, Any]) -> dict[str, str]:
@ -163,9 +88,36 @@ def build_app(album: Album) -> FastAPI:
raise HTTPException(status_code=400, detail=str(error)) from error
return {"status": "ok"}
@app.post("/api/person")
async def person(payload: dict[str, Any]) -> dict[str, str]:
# Mid-session, not only at setup: the reason to add somebody is
# usually a photo of them already on the screen.
try:
key = album.add_person([str(t) for t in payload.get("triggers", [])])
except ConfigError as error:
raise HTTPException(status_code=400, detail=str(error)) from error
return {"status": "ok", "key": key}
def build_app(album: Album) -> FastAPI:
"""The one-page application, bound to one album."""
app = FastAPI()
roster_routes(app, album)
@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/tag")
async def tag(payload: dict[str, Any]) -> dict[str, str]:
album.tag(str(payload["file"]), [str(key) for key in payload.get("keys", [])])
try:
album.tag(str(payload["file"]), [str(key) for key in payload.get("keys", [])])
except KeyError as error:
raise HTTPException(status_code=404, detail=str(error)) from error
return {"status": "ok"}
@app.get("/api/summary")
@ -195,17 +147,27 @@ def main() -> int:
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"
"--import", dest="source", type=Path, default=None, help="import media from here first"
)
parser.add_argument(
"--dry-run", action="store_true", help="report what --import would do, change nothing"
)
parser.add_argument(
"--untagged", action="store_true", help="present only items with no tags yet"
)
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)
directory = arguments.dir or default_dirs()[0][0]
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")
report = _import(arguments.source, directory, dry_run=arguments.dry_run)
sys.stdout.write("\n".join(report.lines()) + "\n\n")
if arguments.dry_run:
# Nothing was written, so there is nothing to tag and no reason to
# open a browser at a server this run is not going to start.
return 0
album = Album(directory, untagged_only=arguments.untagged)
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")
@ -216,14 +178,12 @@ def main() -> int:
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
def _import(source: Path, directory: Path, *, dry_run: bool) -> ImportReport:
"""Import a folder, or work out what importing it would do."""
stripper = ExiftoolStripper()
if dry_run:
return plan_import(source, directory, stripper)
return run_import(source, directory, stripper)
if __name__ == "__main__":