"""Read-only HTTP API for the Arakawa Concierge chat logger. Exposes recent / since / search / list_chats over Tailscale so that the Hermes Agent on the Mac (and any other Tailscale-connected client) can query the SQLite chat log via a small stdio MCP wrapper. - Bound to a Tailscale-only address (no public exposure). - Bearer-token auth (CHATLOGGER_API_TOKEN). - Read-only: never writes to the database. - Implementation uses only the Python standard library (no FastAPI/Flask) to avoid pulling new deps into the bot virtualenv. """ from __future__ import annotations import json import logging import os import sys import time from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any from urllib.parse import parse_qs, urlparse # Reuse Store / formatting from the bot package next to this file. sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from store import Store, StoredMessage, format_messages # noqa: E402 LOG = logging.getLogger("chatlogger.api") CONFIG = { "bind_host": os.environ.get("CHATLOGGER_API_BIND", "100.123.47.7"), "bind_port": int(os.environ.get("CHATLOGGER_API_PORT", "8770")), "token": os.environ.get("CHATLOGGER_API_TOKEN", ""), "db_path": os.environ.get("CHATLOGGER_API_DB", "/opt/chatlogger/chats.db"), "tz_offset_hours": int(os.environ.get("CHATLOGGER_API_TZ_OFFSET_HOURS", "7")), "default_recent_limit": 30, "max_recent_limit": 200, "default_search_limit": 30, "max_search_limit": 100, "max_since_hours": 24 * 30, } def _msg_to_dict(m: StoredMessage) -> dict[str, Any]: return { "id": m.id, "chat_id": m.chat_id, "chat_title": m.chat_title, "message_id": m.message_id, "user_id": m.user_id, "user_display": m.user_display, "text": m.text, "media_type": m.media_type, "media_caption": m.media_caption, "reply_to_message_id": m.reply_to_message_id, "timestamp_utc": m.timestamp_utc, "edited": bool(m.edit_timestamp_utc), } def _resolve_chat_id(store: Store, chat_arg: str | None) -> int | None: """Accept either a numeric chat_id or a (case-insensitive) substring of the title.""" if not chat_arg: return None chat_arg = chat_arg.strip() try: return int(chat_arg) except ValueError: pass needle = chat_arg.lower() for c in store.chats(): title = (c.get("title") or "").lower() if needle and needle in title: return int(c["chat_id"]) return None class Handler(BaseHTTPRequestHandler): server_version = "ChatloggerAPI/1.0" # ---- helpers --------------------------------------------------------- def log_message(self, fmt, *args): LOG.info("%s - %s", self.address_string(), fmt % args) def _send_json(self, status: int, payload: Any) -> None: body = json.dumps(payload, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) def _error(self, status: int, message: str) -> None: self._send_json(status, {"error": message, "status": status}) def _check_auth(self) -> bool: if not CONFIG["token"]: self._error(HTTPStatus.SERVICE_UNAVAILABLE, "server has no token configured") return False provided = self.headers.get("Authorization", "") if not provided.startswith("Bearer "): self._error(HTTPStatus.UNAUTHORIZED, "missing bearer token") return False if provided[len("Bearer "):].strip() != CONFIG["token"]: self._error(HTTPStatus.UNAUTHORIZED, "invalid token") return False return True def _qs(self) -> dict[str, list[str]]: return parse_qs(urlparse(self.path).query, keep_blank_values=False) @staticmethod def _first(qs: dict[str, list[str]], key: str, default: str | None = None) -> str | None: v = qs.get(key) return v[0] if v else default @staticmethod def _int(qs: dict[str, list[str]], key: str, default: int, lo: int, hi: int) -> int: try: v = int(Handler._first(qs, key, str(default)) or default) except (TypeError, ValueError): v = default return max(lo, min(hi, v)) # ---- routing --------------------------------------------------------- def do_GET(self): # noqa: N802 (BaseHTTPRequestHandler API) path = urlparse(self.path).path.rstrip("/") or "/" if path == "/healthz": self._send_json(HTTPStatus.OK, {"status": "ok", "service": "chatlogger-api"}) return if not self._check_auth(): return try: store: Store = self.server.store # type: ignore[attr-defined] qs = self._qs() tz = CONFIG["tz_offset_hours"] if path == "/list_chats": chats = store.chats() return self._send_json(HTTPStatus.OK, { "tz_offset_hours": tz, "chats": [ { "chat_id": c["chat_id"], "title": c.get("title"), "type": c.get("type"), "message_count": c["n"], "last_message_utc": c["last_ts"], } for c in chats ], }) if path == "/recent": chat_arg = self._first(qs, "chat") chat_id = _resolve_chat_id(store, chat_arg) if chat_arg and chat_id is None: return self._error(HTTPStatus.NOT_FOUND, f"no chat matches '{chat_arg}'") limit = self._int(qs, "limit", CONFIG["default_recent_limit"], 1, CONFIG["max_recent_limit"]) msgs = store.recent(chat_id=chat_id, limit=limit) return self._send_json(HTTPStatus.OK, { "tz_offset_hours": tz, "chat_id": chat_id, "count": len(msgs), "messages": [_msg_to_dict(m) for m in msgs], "transcript": format_messages(msgs, tz_offset_hours=tz), }) if path == "/search": query = self._first(qs, "query") or self._first(qs, "q") if not query or not query.strip(): return self._error(HTTPStatus.BAD_REQUEST, "missing 'query' parameter") chat_arg = self._first(qs, "chat") chat_id = _resolve_chat_id(store, chat_arg) if chat_arg and chat_id is None: return self._error(HTTPStatus.NOT_FOUND, f"no chat matches '{chat_arg}'") limit = self._int(qs, "limit", CONFIG["default_search_limit"], 1, CONFIG["max_search_limit"]) try: msgs = store.search(query.strip(), chat_id=chat_id, limit=limit) except Exception as exc: # FTS5 syntax errors etc. return self._error(HTTPStatus.BAD_REQUEST, f"search failed: {exc}") # search returns newest-first; flip to chronological for transcript msgs_chrono = list(reversed(msgs)) return self._send_json(HTTPStatus.OK, { "tz_offset_hours": tz, "chat_id": chat_id, "query": query.strip(), "count": len(msgs), "messages": [_msg_to_dict(m) for m in msgs], "transcript": format_messages(msgs_chrono, tz_offset_hours=tz), }) if path == "/since": chat_arg = self._first(qs, "chat") chat_id = _resolve_chat_id(store, chat_arg) if chat_arg and chat_id is None: return self._error(HTTPStatus.NOT_FOUND, f"no chat matches '{chat_arg}'") hours = self._int(qs, "hours", 12, 1, CONFIG["max_since_hours"]) limit = self._int(qs, "limit", 500, 1, 5000) since_ts = int(time.time()) - hours * 3600 msgs = store.since(chat_id=chat_id, since_ts=since_ts, limit=limit) return self._send_json(HTTPStatus.OK, { "tz_offset_hours": tz, "chat_id": chat_id, "since_utc": since_ts, "hours": hours, "count": len(msgs), "messages": [_msg_to_dict(m) for m in msgs], "transcript": format_messages(msgs, tz_offset_hours=tz), }) if path == "/stats": return self._send_json(HTTPStatus.OK, store.stats()) self._error(HTTPStatus.NOT_FOUND, f"unknown route: {path}") except Exception as exc: # last-resort catch LOG.exception("unhandled error on %s", path) self._error(HTTPStatus.INTERNAL_SERVER_ERROR, f"internal error: {exc}") def main() -> None: logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", ) if not CONFIG["token"]: LOG.error("CHATLOGGER_API_TOKEN is not set; refusing to start") sys.exit(2) if not os.path.exists(CONFIG["db_path"]): LOG.error("DB file not found: %s", CONFIG["db_path"]) sys.exit(2) store = Store(CONFIG["db_path"]) server = ThreadingHTTPServer((CONFIG["bind_host"], CONFIG["bind_port"]), Handler) server.store = store # type: ignore[attr-defined] LOG.info("chatlogger-api listening on %s:%s (db=%s)", CONFIG["bind_host"], CONFIG["bind_port"], CONFIG["db_path"]) try: server.serve_forever() except KeyboardInterrupt: LOG.info("shutting down") server.server_close() if __name__ == "__main__": main()