#!/usr/bin/env python3 # Copyright (c) 2026 Krzysztof Rudnicki. MIT licence. """Manage the bot, in a browser, on this machine only. Health, configuration and the photo album in one page. It replaces the tagger, which served the album half of this on the same port; everything it did is still here, and the routes it did it with now live in ``signal_bot/manage/`` where the coverage gate can see them. Started by hand and stopped again, deliberately. This page can rewrite ``.env`` -- the file that decides which groups the bot answers in and who may add it to one -- and a unit would keep that writable on loopback all day for the sake of a bookmark. ./web.sh # the configured pack ./web.sh --dir ~/pack --port 8140 ./web.sh --import ~/photos # dedup, strip and import, then serve ./web.sh --import ~/photos --dry-run # the report, changing nothing ./web.sh --untagged # only items with no tags yet It prints the album 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 # Importing this sets up .venv if the interpreter running us cannot import # what follows. It has to come first, and it is an import rather than a call # so that nothing here needs an E402 suppression. from venv_bootstrap import REPO_ROOT import argparse import sys import webbrowser from pathlib import Path from typing import TYPE_CHECKING, Final import httpx import uvicorn from signal_bot.config_env import load_config from signal_bot.manage.app import build_app from signal_bot.manage.deps import ManageDeps from signal_bot.media_strip import ExiftoolStripper from signal_bot.pack import default_dirs from signal_bot.pack_album import Album from signal_bot.pack_import import plan_import, run_import from signal_bot.portcheck import assert_port_free if TYPE_CHECKING: from signal_bot.pack_report import ImportReport DEFAULT_PORT: Final = 8140 DIST: Final = REPO_ROOT / "web" / "dist" def _arguments() -> argparse.Namespace: """What the launcher was asked to do.""" 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") return parser.parse_args() def main() -> int: """Serve the management page until it is stopped, then print the summary.""" arguments = _arguments() 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 serve and no reason to # open a browser at a server this run is not going to start. return 0 # Before the browser opens, not after: a stale instance on this port # should be a refusal naming the process, not a tab showing somebody # else's album. signal_bot/__main__.py does the same for the bot. assert_port_free("127.0.0.1", arguments.port) album = Album(directory, untagged_only=arguments.untagged) deps = ManageDeps( config=load_config(), env_path=REPO_ROOT / ".env", dist_dir=DIST, album=album, client=httpx.AsyncClient(), port=arguments.port, ) url = f"http://127.0.0.1:{arguments.port}/" sys.stdout.write(f"== {len(album.files)} item(s) in {directory}\n") 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(deps), 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())