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.

Run it through ``./tag.sh``, which sets up ``.venv`` on first use -- the
dependencies are not in the system interpreter, and a script that tells you to
go and install them yourself is a script that does not own its dependencies.

    ./tag.sh                             # the configured pack
    ./tag.sh --dir ~/pack --port 8140
    ./tag.sh --import ~/photos           # dedup, strip and import, then tag
    ./tag.sh --import ~/photos --dry-run # the report, changing nothing
    ./tag.sh --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
this is where that shows up rather than in the group.
"""

from __future__ import annotations

import argparse
import sys
import webbrowser
from pathlib import Path
from typing import TYPE_CHECKING, 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.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


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 roster_routes(app: FastAPI, album: Album) -> None:
    """The two endpoints that write people.json.

    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]:
        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/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]:
        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")
    async def summary() -> dict[str, list[str]]:
        # The same lines the tagger prints when it stops. Shown in the page
        # too, because "I tagged everything, did it save?" is a question the
        # last photo used to answer by doing nothing at all.
        return {"lines": album.summary()}

    @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="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_dirs()[0][0]
    if arguments.source is not None:
        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")
    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, 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__":
    raise SystemExit(main())