signal-bot/tool/tagger.py

#!/usr/bin/env python3
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Tag a private photo album, in a browser, on this machine only.

Sixty-odd photos and seven people is not a hand-edited JSON file. This serves
one page on 127.0.0.1 that shows each item in turn with a numbered checkbox
per person, saves on every keystroke, and never leaves the machine.

Names are typed *into the page*, never as arguments: an argument lands in the
shell history, and the whole design of the pack is that the names exist in
one directory and nowhere else.

    python tool/tagger.py                      # the configured pack
    python tool/tagger.py --dir ~/pack --port 8140
    python tool/tagger.py --import ~/photos    # copy files in, then tag

It prints a summary when it stops: how many items are untagged, and how many
photos each command would have. A command with zero is dead on arrival, and
this is where that shows up rather than in the group.
"""

from __future__ import annotations

import argparse
import json
import shutil
import sys
import webbrowser
from pathlib import Path
from typing import Any, Final

import httpx
import uvicorn
from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse, HTMLResponse

from signal_bot.config_env import load_config
from signal_bot.groups import ConfigError
from signal_bot.pack import MEDIA_DIR, MEDIA_TYPES, PEOPLE_FILE, TAGS_FILE, default_dir
from signal_bot.pack_load import load

PAGE: Final = Path(__file__).with_name("tagger.html")
DEFAULT_PORT: Final = 8140


class Album:
    """The pack directory, and every read and write the page needs."""

    def __init__(self, directory: Path) -> None:
        """Point the tagger at one pack, creating the album directory."""
        self.directory = directory
        (directory / MEDIA_DIR).mkdir(parents=True, exist_ok=True)

    @property
    def files(self) -> list[str]:
        """Every taggable file, in a stable order."""
        album = self.directory / MEDIA_DIR
        return sorted(path.name for path in album.iterdir() if path.suffix.lower() in MEDIA_TYPES)

    def read(self, name: str) -> dict[str, Any]:
        """One manifest, or an empty one."""
        path = self.directory / name
        if not path.exists():
            return {}
        document: dict[str, Any] = json.loads(path.read_text(encoding="utf-8"))
        return document

    def write(self, name: str, document: dict[str, Any]) -> None:
        """Replace one manifest atomically, so a closed tab loses nothing."""
        path = self.directory / name
        temporary = path.with_suffix(".tmp")
        temporary.write_text(json.dumps(document, indent=2, ensure_ascii=False), encoding="utf-8")
        temporary.replace(path)

    def state(self) -> dict[str, Any]:
        """Everything the page draws itself from."""
        people = self.read(PEOPLE_FILE)
        tags = self.read(TAGS_FILE)
        return {
            "configured": bool(people.get("people")),
            "people": people.get("people", []),
            "combos": people.get("combos", []),
            "groups": people.get("groups", []),
            "files": [{"name": name, "keys": tags.get(name, [])} for name in self.files],
        }

    def setup(
        self, groups: list[str], people: list[dict[str, Any]], combos: list[dict[str, Any]]
    ) -> None:
        """Write people.json, then prove the bot could load it."""
        self.write(PEOPLE_FILE, {"groups": groups, "people": people, "combos": combos})
        load(self.directory)

    def tag(self, filename: str, keys: list[str]) -> None:
        """Record who is in one file."""
        if filename not in self.files:
            raise HTTPException(status_code=404, detail=f"{filename} is not in the album")
        tags = self.read(TAGS_FILE)
        tags[filename] = keys
        self.write(TAGS_FILE, tags)

    def summary(self) -> list[str]:
        """What each command would answer with, and what is still untagged."""
        tags = self.read(TAGS_FILE)
        lines = [f"== {len(self.files)} items, {len(self.files) - len(tags)} never opened"]
        try:
            pack = load(self.directory)
        except ConfigError as error:
            # Reported rather than raised: the summary prints as the tagger
            # exits, and a half-built pack is the normal state to stop in.
            return [*lines, f"!! the pack does not load yet: {error}"]
        for entry in pack.entries:
            counts = [len(pack.matching(option)) for option in entry.options]
            total = sum(counts)
            mark = "  " if total else "!!"
            lines.append(f"{mark} {entry.triggers[0]}: {total} photo(s)")
        return lines


def live_groups() -> list[dict[str, str]]:
    """The groups this account is in, so an id is picked and never typed."""
    try:
        config = load_config()
        response = httpx.get(config.groups_url, timeout=5.0)
        response.raise_for_status()
        listing = response.json()
    except ConfigError, httpx.HTTPError, ValueError:
        # No container, no .env, no network: none of them stop the tagging,
        # they only mean the group id has to be pasted instead of picked.
        return []
    return [
        {"id": str(entry.get("id", "")), "name": str(entry.get("name", ""))}
        for entry in listing
        if entry.get("id")
    ]


def build_app(album: Album) -> FastAPI:
    """The one-page application, bound to one album."""
    app = FastAPI()

    @app.get("/", response_class=HTMLResponse)
    async def page() -> str:
        return PAGE.read_text(encoding="utf-8")

    @app.get("/api/state")
    async def state() -> dict[str, Any]:
        return {**album.state(), "known_groups": live_groups()}

    @app.post("/api/setup")
    async def setup(payload: dict[str, Any]) -> dict[str, str]:
        try:
            album.setup(
                list(payload.get("groups", [])),
                list(payload.get("people", [])),
                list(payload.get("combos", [])),
            )
        except ConfigError as error:
            # Straight into the page: a manifest the bot would refuse should
            # be fixed here, in the form, not found at the next restart.
            raise HTTPException(status_code=400, detail=str(error)) from error
        return {"status": "ok"}

    @app.post("/api/tag")
    async def tag(payload: dict[str, Any]) -> dict[str, str]:
        album.tag(str(payload["file"]), [str(key) for key in payload.get("keys", [])])
        return {"status": "ok"}

    @app.get("/media/{index}")
    async def media(index: int) -> FileResponse:
        # By index, never by a name the page sends back: a filename from a
        # client is a path, and portraits.path_for exists for the same reason.
        files = album.files
        if not 0 <= index < len(files):
            raise HTTPException(status_code=404, detail="no such item")
        return FileResponse(album.directory / MEDIA_DIR / files[index])

    return app


def main() -> int:
    """Serve the tagger until it is stopped, then print the summary."""
    parser = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    parser.add_argument("--dir", type=Path, default=None, help="the pack directory")
    parser.add_argument("--port", type=int, default=DEFAULT_PORT)
    parser.add_argument(
        "--import", dest="source", type=Path, default=None, help="copy media in from here first"
    )
    parser.add_argument("--no-open", action="store_true", help="do not open a browser")
    arguments = parser.parse_args()

    directory = arguments.dir or default_dir()[0]
    album = Album(directory)
    if arguments.source is not None:
        copied = _import(arguments.source, directory / MEDIA_DIR)
        sys.stdout.write(f"== copied {copied} file(s) into {directory / MEDIA_DIR}\n")

    sys.stdout.write(f"== tagging {len(album.files)} item(s) in {directory}\n")
    url = f"http://127.0.0.1:{arguments.port}/"
    sys.stdout.write(f"== open {url} -- ctrl-c here when you are done\n")
    if not arguments.no_open:
        webbrowser.open(url)
    uvicorn.run(build_app(album), host="127.0.0.1", port=arguments.port, log_level="warning")
    sys.stdout.write("\n".join(album.summary()) + "\n")
    return 0


def _import(source: Path, album: Path) -> int:
    """Copy, never move: the originals stay put until the tagging is done."""
    copied = 0
    for path in sorted(source.iterdir()):
        if path.suffix.lower() in MEDIA_TYPES and not (album / path.name).exists():
            shutil.copy2(path, album / path.name)
            copied += 1
    return copied


if __name__ == "__main__":
    raise SystemExit(main())