signal-bot/signal_bot/pack_album.py

# 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