feat(chatlogger): Arakawa Concierge — passive Telegram group chat logger

Standalone bot @arakawa_concierge_bot that silently records configured
group chats into SQLite (FTS5). Phase 1 — capture only, no LLM, no
auto-replies. Reads token + chat allowlist from /opt/chatlogger/env
(NOT in git). Runs as systemd unit chatlogger-bot.service on CT 116
sharing /opt/bot-venv with the hausmeister-bot.

- chatlogger_bot.py: telegram polling, message/edit handlers, admin cmds
  (/whichchat, /log_status, /recent, /search) — silent in groups
- store.py: SQLite schema + FTS5 search + edit audit
- query.py: CLI (recent, search, stats, chats, since, export JSONL)
- chatlogger-bot.service: systemd unit, hardened (ProtectSystem=strict)
- env.example, README.md

Tested live: message 'hallo' from -1003924901022 stored, retrieved
via CLI within seconds. Group privacy must be off in BotFather.

chatlogger/README.md

@@ -0,0 +1,109 @@
+# Arakawa Concierge — Group Chat Logger
+
+Stand-alone Telegram bot (`@arakawa_concierge_bot`) that silently records
+configured group chats into a SQLite database. Intended as the read-only
+data source for the Hausmeister bot (Jarvis) so it can answer questions
+about what was said, agreed, or left open in a managed chat.
+
+**Phase 1 (this code):** capture only — no auto-replies, no LLM, no
+classification. Only the admin can use admin commands; in groups the bot
+is mute except when the admin sends `/whichchat`.
+
+## Files
+
+| File | Purpose |
+|------|---------|
+| `chatlogger_bot.py` | Main bot (long-poll, message handlers, admin commands) |
+| `store.py` | SQLite schema + CRUD + FTS5 search |
+| `query.py` | CLI: `recent`, `search`, `stats`, `chats`, `since`, `export` |
+| `chatlogger-bot.service` | systemd unit (installed on CT 116) |
+| `env.example` | template for `/opt/chatlogger/env` (NOT committed) |
+
+## Runtime layout on CT 116
+
+```
+/opt/homelab-brain/chatlogger/   # code (read-only via bind-mount mp1)
+/opt/chatlogger/
+  ├── env                        # token + chat allowlist (chmod 600, root)
+  └── chats.db                   # SQLite, WAL mode
+/etc/systemd/system/chatlogger-bot.service
+```
+
+`/opt/bot-venv` is shared with the hausmeister-bot (python-telegram-bot 22.6).
+
+## First-time install (on CT 116)
+
+```bash
+mkdir -p /opt/chatlogger
+cp /opt/homelab-brain/chatlogger/env.example /opt/chatlogger/env
+$EDITOR /opt/chatlogger/env       # paste token, set chat ids
+chmod 600 /opt/chatlogger/env
+
+cp /opt/homelab-brain/chatlogger/chatlogger-bot.service /etc/systemd/system/
+systemctl daemon-reload
+systemctl enable --now chatlogger-bot
+journalctl -u chatlogger-bot -f
+```
+
+## Adding a new group
+
+1. Add `@arakawa_concierge_bot` to the group as a regular member (no admin needed).
+2. As admin, send `/whichchat` in the group → bot replies with the `chat_id`.
+3. Append the id to `ARAKAWA_BOT_ALLOWED_CHATS` in `/opt/chatlogger/env` (comma separated).
+4. `systemctl restart chatlogger-bot`.
+
+**BotFather group privacy MUST be off** otherwise the bot only sees `@mentions`:
+`/mybots` → `@arakawa_concierge_bot` → `Bot Settings` → `Group Privacy` → `Turn off`,
+then remove and re-add the bot to the group (Telegram caches the privacy flag per chat).
+
+## CLI
+
+```bash
+# Most recent 50 messages, all chats
+/opt/bot-venv/bin/python /opt/homelab-brain/chatlogger/query.py recent -n 50
+
+# Search
+/opt/bot-venv/bin/python /opt/homelab-brain/chatlogger/query.py search "kaution"
+
+# Last 24h of one chat
+/opt/bot-venv/bin/python /opt/homelab-brain/chatlogger/query.py since 1440 --chat -1003924901022
+
+# Per-chat stats
+/opt/bot-venv/bin/python /opt/homelab-brain/chatlogger/query.py stats
+
+# Export as JSONL (e.g. for handing to the Hausmeister-Bot)
+/opt/bot-venv/bin/python /opt/homelab-brain/chatlogger/query.py export --chat -1003924901022 > /tmp/log.jsonl
+```
+
+A wrapper `/usr/local/bin/chatlogger` may be installed for convenience:
+
+```bash
+#!/bin/bash
+exec /opt/bot-venv/bin/python /opt/homelab-brain/chatlogger/query.py "$@"
+```
+
+## Telegram commands (admin only)
+
+| Command | Where | Effect |
+|---------|-------|--------|
+| `/whichchat` | any chat (admin) | Bot replies with `chat_id`, type, title, allowlist status |
+| `/log_status` | any chat (admin) | DB size, message counts per chat |
+| `/recent [N]` | any chat (admin) | Last N messages of the current chat (or all in PM) |
+| `/search <text>` | any chat (admin) | FTS5 search in the current chat (or all in PM) |
+
+These are the *only* messages the bot ever sends. Group members other than
+the admin get no reply at all.
+
+## Schema (SQLite)
+
+`messages` (one row per Telegram message), `messages_fts` (FTS5 mirror),
+`message_edits` (audit log of edits). Unique key `(chat_id, message_id)`
+makes ingestion idempotent — a Telegram retry never duplicates rows.
+
+## Phase 2 (later) — Jarvis access
+
+Add `homelab-ai-bot/tools/chatlog.py` with three tool functions
+(`chatlog_recent`, `chatlog_search`, `chatlog_summary`) that import
+`chatlogger.store` and read the same SQLite file directly (CT 116 sees it
+locally, no API needed). Wire them into `llm.py` routing on keywords
+like „Gruppe", „Airbnb", chat title, etc.

chatlogger/chatlogger-bot.service

@@ -0,0 +1,25 @@
+[Unit]
+Description=Arakawa Concierge — passive Telegram group chat logger
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=simple
+WorkingDirectory=/opt/homelab-brain/chatlogger
+EnvironmentFile=/opt/chatlogger/env
+Environment=PYTHONUNBUFFERED=1
+ExecStart=/opt/bot-venv/bin/python /opt/homelab-brain/chatlogger/chatlogger_bot.py
+Restart=always
+RestartSec=10
+KillMode=control-group
+TimeoutStopSec=10
+
+# Hardening — bot only needs to read repo + write its own data dir.
+ProtectSystem=strict
+ReadWritePaths=/opt/chatlogger
+ProtectHome=true
+PrivateTmp=true
+NoNewPrivileges=true
+
+[Install]
+WantedBy=multi-user.target

chatlogger/chatlogger_bot.py

@@ -0,0 +1,385 @@
+"""Arakawa Concierge — passive Telegram group chat logger.
+
+Reads messages from configured groups, stores them in SQLite,
+never replies to group messages. Provides /whichchat and /log_status
+commands for the admin (private chat or in any chat).
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+import signal
+import sys
+import time
+from typing import Any
+
+from telegram import Update
+from telegram.constants import ChatType
+from telegram.ext import (
+    Application,
+    CommandHandler,
+    ContextTypes,
+    MessageHandler,
+    filters,
+)
+
+from store import Store, format_messages
+
+LOG = logging.getLogger("chatlogger")
+
+
+def _env(name: str, default: str | None = None) -> str | None:
+    val = os.environ.get(name, default)
+    return val.strip() if isinstance(val, str) else val
+
+
+def _env_int(name: str, default: int = 0) -> int:
+    val = _env(name)
+    try:
+        return int(val) if val else default
+    except (TypeError, ValueError):
+        return default
+
+
+def _allowed_chats() -> set[int]:
+    raw = _env("ARAKAWA_BOT_ALLOWED_CHATS", "") or ""
+    out: set[int] = set()
+    for part in raw.split(","):
+        part = part.strip()
+        if not part:
+            continue
+        try:
+            out.add(int(part))
+        except ValueError:
+            LOG.warning("Ignoring non-integer chat id in ALLOWED_CHATS: %r", part)
+    return out
+
+
+CONFIG = {
+    "token": _env("ARAKAWA_BOT_TOKEN"),
+    "db_path": _env("ARAKAWA_BOT_DB", "/opt/chatlogger/chats.db"),
+    "admin_user_id": _env_int("ARAKAWA_BOT_ADMIN_USER_ID", 0),
+    "tz_offset_hours": _env_int("ARAKAWA_BOT_TZ_OFFSET_HOURS", 7),
+}
+
+
+def _extract_forward_origin(message) -> str | None:
+    """Return a short string describing where a forwarded message came from.
+
+    python-telegram-bot 22.x uses MessageOrigin* classes via message.forward_origin.
+    Keep this defensive — older fields may also exist depending on PTB version.
+    """
+    origin = getattr(message, "forward_origin", None)
+    if origin is not None:
+        sender_user = getattr(origin, "sender_user", None)
+        if sender_user is not None:
+            return getattr(sender_user, "full_name", None) or getattr(sender_user, "username", None)
+        sender_chat = getattr(origin, "sender_chat", None) or getattr(origin, "chat", None)
+        if sender_chat is not None:
+            return getattr(sender_chat, "title", None) or getattr(sender_chat, "username", None)
+        sender_user_name = getattr(origin, "sender_user_name", None)
+        if sender_user_name:
+            return sender_user_name
+        return type(origin).__name__
+    legacy_user = getattr(message, "forward_from", None)
+    if legacy_user is not None:
+        return getattr(legacy_user, "full_name", None)
+    legacy_chat = getattr(message, "forward_from_chat", None)
+    if legacy_chat is not None:
+        return getattr(legacy_chat, "title", None) or getattr(legacy_chat, "username", None)
+    return None
+
+
+def detect_media_type(message) -> tuple[str | None, str | None]:
+    """Return (media_type, file_id_or_summary) for the message, if any."""
+    if message.photo:
+        return "photo", message.photo[-1].file_id
+    if message.document:
+        return "document", message.document.file_id
+    if message.voice:
+        return "voice", message.voice.file_id
+    if message.video:
+        return "video", message.video.file_id
+    if message.video_note:
+        return "video_note", message.video_note.file_id
+    if message.audio:
+        return "audio", message.audio.file_id
+    if message.sticker:
+        return "sticker", message.sticker.file_id
+    if message.animation:
+        return "animation", message.animation.file_id
+    if message.location:
+        loc = message.location
+        return "location", f"{loc.latitude},{loc.longitude}"
+    if message.venue:
+        v = message.venue
+        return "venue", f"{v.title} | {v.address}"
+    if message.contact:
+        return "contact", message.contact.phone_number or ""
+    if message.poll:
+        return "poll", message.poll.question
+    if message.dice:
+        return "dice", str(message.dice.value)
+    return None, None
+
+
+async def on_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    msg = update.effective_message
+    chat = update.effective_chat
+    if msg is None or chat is None:
+        return
+
+    allowed = context.bot_data["allowed_chats"]
+    if chat.id not in allowed:
+        return
+
+    user = msg.from_user
+    media_type, media_payload = detect_media_type(msg)
+
+    forward_from = _extract_forward_origin(msg)
+
+    is_edit = bool(update.edited_message)
+    store: Store = context.bot_data["store"]
+
+    try:
+        raw_json = msg.to_json()
+    except Exception:
+        raw_json = "{}"
+
+    if is_edit:
+        edit_ts = int(msg.edit_date.timestamp()) if msg.edit_date else int(time.time())
+        ok = store.update_message(
+            chat_id=chat.id,
+            message_id=msg.message_id,
+            new_text=msg.text,
+            new_caption=msg.caption,
+            edit_timestamp_utc=edit_ts,
+        )
+        if not ok:
+            store.add_message(
+                chat_id=chat.id,
+                chat_title=chat.title,
+                chat_type=str(chat.type),
+                message_id=msg.message_id,
+                user_id=user.id if user else None,
+                user_first_name=user.first_name if user else None,
+                user_last_name=user.last_name if user else None,
+                user_username=user.username if user else None,
+                text=msg.text,
+                media_type=media_type,
+                media_caption=msg.caption,
+                media_file_id=media_payload,
+                reply_to_message_id=msg.reply_to_message.message_id if msg.reply_to_message else None,
+                forward_from=forward_from,
+                timestamp_utc=edit_ts,
+                raw_json=raw_json,
+            )
+        LOG.info("EDIT  chat=%s msg=%s user=%s", chat.id, msg.message_id, user.id if user else None)
+        return
+
+    ts = int(msg.date.timestamp()) if msg.date else int(time.time())
+    inserted = store.add_message(
+        chat_id=chat.id,
+        chat_title=chat.title,
+        chat_type=str(chat.type),
+        message_id=msg.message_id,
+        user_id=user.id if user else None,
+        user_first_name=user.first_name if user else None,
+        user_last_name=user.last_name if user else None,
+        user_username=user.username if user else None,
+        text=msg.text,
+        media_type=media_type,
+        media_caption=msg.caption,
+        media_file_id=media_payload,
+        reply_to_message_id=msg.reply_to_message.message_id if msg.reply_to_message else None,
+        forward_from=forward_from,
+        timestamp_utc=ts,
+        raw_json=raw_json,
+    )
+    LOG.info(
+        "%s chat=%s msg=%s user=%s media=%s text=%r",
+        "STORE" if inserted else "DUP  ",
+        chat.id,
+        msg.message_id,
+        user.id if user else None,
+        media_type,
+        (msg.text or msg.caption or "")[:80],
+    )
+
+
+async def cmd_whichchat(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    """Reply with current chat id + title. Open to anyone — useful for setup.
+
+    The reply itself is the only message this bot ever sends in a group chat.
+    Restricted to admin user to avoid noise from random people typing /whichchat.
+    """
+    chat = update.effective_chat
+    user = update.effective_user
+    if chat is None or user is None:
+        return
+    admin = context.bot_data["admin_user_id"]
+    if admin and user.id != admin:
+        return
+
+    allowed = context.bot_data["allowed_chats"]
+    status = "AKTIV (wird geloggt)" if chat.id in allowed else "NICHT konfiguriert"
+    text = (
+        f"chat_id: <code>{chat.id}</code>\n"
+        f"type: {chat.type}\n"
+        f"title: {chat.title or '(privat)'}\n"
+        f"status: {status}"
+    )
+    await context.bot.send_message(chat_id=chat.id, text=text, parse_mode="HTML")
+
+
+async def cmd_log_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    user = update.effective_user
+    chat = update.effective_chat
+    if chat is None or user is None:
+        return
+    admin = context.bot_data["admin_user_id"]
+    if admin and user.id != admin:
+        return
+
+    store: Store = context.bot_data["store"]
+    s = store.stats()
+    lines = [
+        f"📒 <b>Arakawa Concierge — Status</b>",
+        f"DB: <code>{store.db_path}</code> ({s['db_size_bytes'] / 1024:.1f} KB)",
+        f"Nachrichten gesamt: <b>{s['total_messages']}</b>",
+        f"Edits: {s['total_edits']}",
+        "",
+        "<b>Pro Chat:</b>",
+    ]
+    for c in s["chats"]:
+        last = time.strftime("%Y-%m-%d %H:%M", time.gmtime(c["last_ts"] + CONFIG["tz_offset_hours"] * 3600))
+        lines.append(f"  • <code>{c['chat_id']}</code> {c['title'] or '?'}: {c['n']} (letzter {last})")
+    if not s["chats"]:
+        lines.append("  (noch keine Nachrichten)")
+    await context.bot.send_message(chat_id=chat.id, text="\n".join(lines), parse_mode="HTML")
+
+
+async def cmd_recent(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    user = update.effective_user
+    chat = update.effective_chat
+    if chat is None or user is None:
+        return
+    admin = context.bot_data["admin_user_id"]
+    if admin and user.id != admin:
+        return
+
+    args = context.args or []
+    n = 20
+    if args:
+        try:
+            n = max(1, min(200, int(args[0])))
+        except ValueError:
+            pass
+    store: Store = context.bot_data["store"]
+    target_chat = chat.id if chat.type != ChatType.PRIVATE else None
+    msgs = store.recent(chat_id=target_chat, limit=n)
+    if not msgs:
+        await context.bot.send_message(chat_id=chat.id, text="(keine Nachrichten gespeichert)")
+        return
+    text = format_messages(msgs, tz_offset_hours=CONFIG["tz_offset_hours"])
+    for chunk in _chunks(text, 3500):
+        await context.bot.send_message(chat_id=chat.id, text=f"<pre>{_html_escape(chunk)}</pre>", parse_mode="HTML")
+
+
+async def cmd_search(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    user = update.effective_user
+    chat = update.effective_chat
+    if chat is None or user is None:
+        return
+    admin = context.bot_data["admin_user_id"]
+    if admin and user.id != admin:
+        return
+
+    args = context.args or []
+    if not args:
+        await context.bot.send_message(chat_id=chat.id, text="Nutzung: /search &lt;begriff&gt;", parse_mode="HTML")
+        return
+    query = " ".join(args)
+    store: Store = context.bot_data["store"]
+    target_chat = chat.id if chat.type != ChatType.PRIVATE else None
+    msgs = store.search(query=query, chat_id=target_chat, limit=30)
+    if not msgs:
+        await context.bot.send_message(chat_id=chat.id, text=f"(keine Treffer für „{query}“)")
+        return
+    text = format_messages(msgs, tz_offset_hours=CONFIG["tz_offset_hours"])
+    for chunk in _chunks(text, 3500):
+        await context.bot.send_message(chat_id=chat.id, text=f"<pre>{_html_escape(chunk)}</pre>", parse_mode="HTML")
+
+
+def _chunks(text: str, max_len: int):
+    if len(text) <= max_len:
+        yield text
+        return
+    cur = []
+    cur_len = 0
+    for line in text.split("\n"):
+        if cur_len + len(line) + 1 > max_len and cur:
+            yield "\n".join(cur)
+            cur = [line]
+            cur_len = len(line) + 1
+        else:
+            cur.append(line)
+            cur_len += len(line) + 1
+    if cur:
+        yield "\n".join(cur)
+
+
+def _html_escape(s: str) -> str:
+    return s.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
+
+
+def build_app() -> Application:
+    if not CONFIG["token"]:
+        raise SystemExit("ARAKAWA_BOT_TOKEN missing in environment")
+
+    store = Store(CONFIG["db_path"])
+    allowed = _allowed_chats()
+    LOG.info(
+        "Starting chatlogger: db=%s allowed_chats=%s admin=%s",
+        CONFIG["db_path"], sorted(allowed), CONFIG["admin_user_id"],
+    )
+
+    app = Application.builder().token(CONFIG["token"]).build()
+    app.bot_data["store"] = store
+    app.bot_data["allowed_chats"] = allowed
+    app.bot_data["admin_user_id"] = CONFIG["admin_user_id"]
+
+    app.add_handler(CommandHandler("whichchat", cmd_whichchat))
+    app.add_handler(CommandHandler("log_status", cmd_log_status))
+    app.add_handler(CommandHandler("recent", cmd_recent))
+    app.add_handler(CommandHandler("search", cmd_search))
+
+    msg_filter = (
+        filters.ChatType.GROUPS | filters.ChatType.SUPERGROUP | filters.ChatType.PRIVATE
+    ) & ~filters.COMMAND
+    app.add_handler(MessageHandler(msg_filter, on_message))
+    app.add_handler(MessageHandler(filters.UpdateType.EDITED_MESSAGE & ~filters.COMMAND, on_message))
+
+    return app
+
+
+def main() -> None:
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+        stream=sys.stdout,
+    )
+    logging.getLogger("httpx").setLevel(logging.WARNING)
+    logging.getLogger("telegram").setLevel(logging.WARNING)
+
+    app = build_app()
+    app.run_polling(
+        allowed_updates=["message", "edited_message", "channel_post", "my_chat_member"],
+        drop_pending_updates=False,
+    )
+
+
+if __name__ == "__main__":
+    main()

chatlogger/env.example

@@ -0,0 +1,20 @@
+# /opt/chatlogger/env  —  install on CT 116 with chmod 600
+# NOT committed to git. Loaded via systemd EnvironmentFile.
+
+# Telegram bot token from BotFather (@arakawa_concierge_bot)
+ARAKAWA_BOT_TOKEN=PASTE_TOKEN_HERE
+
+# Comma-separated list of chat ids the bot is allowed to log.
+# Anything else is silently ignored. Get the id by adding the bot to a chat
+# and (as admin) sending /whichchat.
+ARAKAWA_BOT_ALLOWED_CHATS=-1003924901022
+
+# Telegram user id allowed to use admin commands (/log_status, /recent, /search).
+# Same id used for the Hausmeister bot in homelab.conf as TG_CHAT_ID.
+ARAKAWA_BOT_ADMIN_USER_ID=674951792
+
+# Time zone offset in hours (Phnom Penh = +7). Used for human-readable timestamps.
+ARAKAWA_BOT_TZ_OFFSET_HOURS=7
+
+# SQLite path — only this directory is writable for the systemd unit.
+ARAKAWA_BOT_DB=/opt/chatlogger/chats.db

chatlogger/query.py

@@ -0,0 +1,143 @@
+"""CLI access to the chatlogger SQLite database.
+
+Usage:
+    chatlogger recent [--chat <id>] [-n 50]
+    chatlogger search "query string" [--chat <id>] [-n 30]
+    chatlogger stats
+    chatlogger chats
+    chatlogger since <minutes> [--chat <id>]
+    chatlogger export [--chat <id>] [--since-ts N]   # JSONL to stdout
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+import time
+
+from store import Store, format_messages
+
+DEFAULT_DB = os.environ.get("ARAKAWA_BOT_DB", "/opt/chatlogger/chats.db")
+TZ_OFFSET_HOURS = int(os.environ.get("ARAKAWA_BOT_TZ_OFFSET_HOURS", "7"))
+
+
+def cmd_recent(args, store: Store) -> int:
+    msgs = store.recent(chat_id=args.chat, limit=args.n)
+    if not msgs:
+        print("(no messages)")
+        return 0
+    print(format_messages(msgs, tz_offset_hours=TZ_OFFSET_HOURS))
+    return 0
+
+
+def cmd_search(args, store: Store) -> int:
+    msgs = store.search(query=args.query, chat_id=args.chat, limit=args.n)
+    if not msgs:
+        print(f"(no matches for {args.query!r})")
+        return 0
+    print(format_messages(msgs, tz_offset_hours=TZ_OFFSET_HOURS))
+    return 0
+
+
+def cmd_stats(args, store: Store) -> int:
+    s = store.stats()
+    print(f"Total messages: {s['total_messages']}")
+    print(f"Total edits:    {s['total_edits']}")
+    print(f"DB size:        {s['db_size_bytes'] / 1024:.1f} KB")
+    print()
+    print("Per chat:")
+    for c in s["chats"]:
+        last = time.strftime("%Y-%m-%d %H:%M", time.gmtime(c["last_ts"] + TZ_OFFSET_HOURS * 3600))
+        print(f"  {c['chat_id']:>20}  {c['n']:>6}  last={last}  title={c['title'] or '?'}")
+    return 0
+
+
+def cmd_chats(args, store: Store) -> int:
+    rows = store.chats()
+    if not rows:
+        print("(no chats)")
+        return 0
+    for c in rows:
+        last = time.strftime("%Y-%m-%d %H:%M", time.gmtime(c["last_ts"] + TZ_OFFSET_HOURS * 3600))
+        print(f"{c['chat_id']:>20}  {c['type']:<12}  count={c['n']:<5}  last={last}  {c['title'] or ''}")
+    return 0
+
+
+def cmd_since(args, store: Store) -> int:
+    since_ts = int(time.time()) - args.minutes * 60
+    msgs = store.since(chat_id=args.chat, since_ts=since_ts, limit=10_000)
+    if not msgs:
+        print("(no messages in window)")
+        return 0
+    print(format_messages(msgs, tz_offset_hours=TZ_OFFSET_HOURS))
+    return 0
+
+
+def cmd_export(args, store: Store) -> int:
+    if args.since_ts:
+        msgs = store.since(chat_id=args.chat, since_ts=args.since_ts, limit=1_000_000)
+    else:
+        msgs = store.recent(chat_id=args.chat, limit=1_000_000)
+    for m in msgs:
+        print(json.dumps({
+            "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,
+            "edit_timestamp_utc": m.edit_timestamp_utc,
+        }, ensure_ascii=False))
+    return 0
+
+
+def build_parser() -> argparse.ArgumentParser:
+    p = argparse.ArgumentParser(prog="chatlogger", description="Arakawa Concierge chat log query tool")
+    p.add_argument("--db", default=DEFAULT_DB, help=f"SQLite path (default: {DEFAULT_DB})")
+    sub = p.add_subparsers(dest="cmd", required=True)
+
+    r = sub.add_parser("recent", help="Last N messages (chronological)")
+    r.add_argument("--chat", type=int, default=None)
+    r.add_argument("-n", type=int, default=50)
+    r.set_defaults(func=cmd_recent)
+
+    s = sub.add_parser("search", help="FTS5 search")
+    s.add_argument("query")
+    s.add_argument("--chat", type=int, default=None)
+    s.add_argument("-n", type=int, default=30)
+    s.set_defaults(func=cmd_search)
+
+    st = sub.add_parser("stats", help="DB statistics")
+    st.set_defaults(func=cmd_stats)
+
+    ch = sub.add_parser("chats", help="List known chats")
+    ch.set_defaults(func=cmd_chats)
+
+    si = sub.add_parser("since", help="Messages from the last N minutes")
+    si.add_argument("minutes", type=int)
+    si.add_argument("--chat", type=int, default=None)
+    si.set_defaults(func=cmd_since)
+
+    ex = sub.add_parser("export", help="Export as JSONL")
+    ex.add_argument("--chat", type=int, default=None)
+    ex.add_argument("--since-ts", type=int, default=None)
+    ex.set_defaults(func=cmd_export)
+
+    return p
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = build_parser()
+    args = parser.parse_args(argv)
+    store = Store(args.db)
+    return args.func(args, store)
+
+
+if __name__ == "__main__":
+    sys.exit(main())

chatlogger/store.py

@@ -0,0 +1,290 @@
+"""SQLite store for the Arakawa Concierge chat logger.
+
+One DB file holds all watched groups. Schema is created on first open.
+Full-text search via FTS5 across message text, captions, sender names.
+"""
+from __future__ import annotations
+
+import json
+import os
+import sqlite3
+import time
+from contextlib import contextmanager
+from dataclasses import dataclass
+from typing import Any, Iterable
+
+SCHEMA = """
+CREATE TABLE IF NOT EXISTS messages (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    chat_id INTEGER NOT NULL,
+    chat_title TEXT,
+    chat_type TEXT,
+    message_id INTEGER NOT NULL,
+    user_id INTEGER,
+    user_first_name TEXT,
+    user_last_name TEXT,
+    user_username TEXT,
+    text TEXT,
+    media_type TEXT,
+    media_caption TEXT,
+    media_file_id TEXT,
+    reply_to_message_id INTEGER,
+    forward_from TEXT,
+    timestamp_utc INTEGER NOT NULL,
+    edit_timestamp_utc INTEGER,
+    raw_json TEXT,
+    UNIQUE(chat_id, message_id)
+);
+
+CREATE INDEX IF NOT EXISTS idx_chat_ts ON messages(chat_id, timestamp_utc);
+CREATE INDEX IF NOT EXISTS idx_user_ts ON messages(user_id, timestamp_utc);
+
+CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts5(
+    text, media_caption, user_first_name, user_username,
+    content='messages', content_rowid='id',
+    tokenize='unicode61 remove_diacritics 1'
+);
+
+CREATE TRIGGER IF NOT EXISTS messages_ai AFTER INSERT ON messages BEGIN
+    INSERT INTO messages_fts(rowid, text, media_caption, user_first_name, user_username)
+    VALUES (new.id, new.text, new.media_caption, new.user_first_name, new.user_username);
+END;
+
+CREATE TRIGGER IF NOT EXISTS messages_ad AFTER DELETE ON messages BEGIN
+    INSERT INTO messages_fts(messages_fts, rowid, text, media_caption, user_first_name, user_username)
+    VALUES ('delete', old.id, old.text, old.media_caption, old.user_first_name, old.user_username);
+END;
+
+CREATE TRIGGER IF NOT EXISTS messages_au AFTER UPDATE ON messages BEGIN
+    INSERT INTO messages_fts(messages_fts, rowid, text, media_caption, user_first_name, user_username)
+    VALUES ('delete', old.id, old.text, old.media_caption, old.user_first_name, old.user_username);
+    INSERT INTO messages_fts(rowid, text, media_caption, user_first_name, user_username)
+    VALUES (new.id, new.text, new.media_caption, new.user_first_name, new.user_username);
+END;
+
+CREATE TABLE IF NOT EXISTS message_edits (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    chat_id INTEGER NOT NULL,
+    message_id INTEGER NOT NULL,
+    text_before TEXT,
+    text_after TEXT,
+    edit_timestamp_utc INTEGER NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS idx_edits_msg ON message_edits(chat_id, message_id);
+"""
+
+
+@dataclass
+class StoredMessage:
+    id: int
+    chat_id: int
+    chat_title: str | None
+    message_id: int
+    user_id: int | None
+    user_display: str
+    text: str | None
+    media_type: str | None
+    media_caption: str | None
+    reply_to_message_id: int | None
+    timestamp_utc: int
+    edit_timestamp_utc: int | None
+
+
+class Store:
+    def __init__(self, db_path: str):
+        self.db_path = db_path
+        os.makedirs(os.path.dirname(db_path) or ".", exist_ok=True)
+        with self._conn() as c:
+            c.executescript(SCHEMA)
+            c.execute("PRAGMA journal_mode=WAL")
+            c.execute("PRAGMA synchronous=NORMAL")
+
+    @contextmanager
+    def _conn(self):
+        conn = sqlite3.connect(self.db_path, timeout=10, isolation_level=None)
+        conn.row_factory = sqlite3.Row
+        try:
+            yield conn
+        finally:
+            conn.close()
+
+    @staticmethod
+    def _user_display(row: sqlite3.Row | dict[str, Any]) -> str:
+        first = (row["user_first_name"] or "").strip()
+        last = (row["user_last_name"] or "").strip()
+        uname = (row["user_username"] or "").strip()
+        full = " ".join(p for p in (first, last) if p)
+        if full and uname:
+            return f"{full} (@{uname})"
+        return full or (f"@{uname}" if uname else f"user:{row['user_id']}")
+
+    def add_message(
+        self,
+        *,
+        chat_id: int,
+        chat_title: str | None,
+        chat_type: str | None,
+        message_id: int,
+        user_id: int | None,
+        user_first_name: str | None,
+        user_last_name: str | None,
+        user_username: str | None,
+        text: str | None,
+        media_type: str | None,
+        media_caption: str | None,
+        media_file_id: str | None,
+        reply_to_message_id: int | None,
+        forward_from: str | None,
+        timestamp_utc: int,
+        raw_json: str,
+    ) -> bool:
+        with self._conn() as c:
+            try:
+                c.execute(
+                    """INSERT INTO messages (
+                        chat_id, chat_title, chat_type, message_id,
+                        user_id, user_first_name, user_last_name, user_username,
+                        text, media_type, media_caption, media_file_id,
+                        reply_to_message_id, forward_from, timestamp_utc, raw_json
+                    ) VALUES (?,?,?,?, ?,?,?,?, ?,?,?,?, ?,?,?,?)""",
+                    (
+                        chat_id, chat_title, chat_type, message_id,
+                        user_id, user_first_name, user_last_name, user_username,
+                        text, media_type, media_caption, media_file_id,
+                        reply_to_message_id, forward_from, timestamp_utc, raw_json,
+                    ),
+                )
+                return True
+            except sqlite3.IntegrityError:
+                return False
+
+    def update_message(
+        self,
+        *,
+        chat_id: int,
+        message_id: int,
+        new_text: str | None,
+        new_caption: str | None,
+        edit_timestamp_utc: int,
+    ) -> bool:
+        with self._conn() as c:
+            row = c.execute(
+                "SELECT id, text, media_caption FROM messages WHERE chat_id=? AND message_id=?",
+                (chat_id, message_id),
+            ).fetchone()
+            if not row:
+                return False
+            text_before = row["text"] or row["media_caption"]
+            text_after = new_text if new_text is not None else new_caption
+            c.execute(
+                """UPDATE messages SET text=?, media_caption=?, edit_timestamp_utc=?
+                   WHERE chat_id=? AND message_id=?""",
+                (new_text, new_caption, edit_timestamp_utc, chat_id, message_id),
+            )
+            c.execute(
+                """INSERT INTO message_edits (chat_id, message_id, text_before, text_after, edit_timestamp_utc)
+                   VALUES (?,?,?,?,?)""",
+                (chat_id, message_id, text_before, text_after, edit_timestamp_utc),
+            )
+            return True
+
+    def recent(self, chat_id: int | None = None, limit: int = 50) -> list[StoredMessage]:
+        with self._conn() as c:
+            if chat_id is None:
+                rows = c.execute(
+                    "SELECT * FROM messages ORDER BY timestamp_utc DESC LIMIT ?",
+                    (limit,),
+                ).fetchall()
+            else:
+                rows = c.execute(
+                    "SELECT * FROM messages WHERE chat_id=? ORDER BY timestamp_utc DESC LIMIT ?",
+                    (chat_id, limit),
+                ).fetchall()
+        return [self._row_to_msg(r) for r in reversed(rows)]
+
+    def since(self, chat_id: int | None, since_ts: int, limit: int = 1000) -> list[StoredMessage]:
+        with self._conn() as c:
+            if chat_id is None:
+                rows = c.execute(
+                    "SELECT * FROM messages WHERE timestamp_utc >= ? ORDER BY timestamp_utc ASC LIMIT ?",
+                    (since_ts, limit),
+                ).fetchall()
+            else:
+                rows = c.execute(
+                    "SELECT * FROM messages WHERE chat_id=? AND timestamp_utc >= ? ORDER BY timestamp_utc ASC LIMIT ?",
+                    (chat_id, since_ts, limit),
+                ).fetchall()
+        return [self._row_to_msg(r) for r in rows]
+
+    def search(self, query: str, chat_id: int | None = None, limit: int = 30) -> list[StoredMessage]:
+        with self._conn() as c:
+            sql = (
+                "SELECT m.* FROM messages m "
+                "JOIN messages_fts f ON f.rowid = m.id "
+                "WHERE messages_fts MATCH ?"
+            )
+            params: list[Any] = [query]
+            if chat_id is not None:
+                sql += " AND m.chat_id = ?"
+                params.append(chat_id)
+            sql += " ORDER BY m.timestamp_utc DESC LIMIT ?"
+            params.append(limit)
+            rows = c.execute(sql, params).fetchall()
+        return [self._row_to_msg(r) for r in rows]
+
+    def stats(self) -> dict[str, Any]:
+        with self._conn() as c:
+            total = c.execute("SELECT COUNT(*) AS n FROM messages").fetchone()["n"]
+            per_chat = c.execute(
+                """SELECT chat_id, MAX(chat_title) AS title, COUNT(*) AS n,
+                          MIN(timestamp_utc) AS first_ts, MAX(timestamp_utc) AS last_ts
+                   FROM messages GROUP BY chat_id ORDER BY n DESC"""
+            ).fetchall()
+            edits = c.execute("SELECT COUNT(*) AS n FROM message_edits").fetchone()["n"]
+        return {
+            "total_messages": total,
+            "total_edits": edits,
+            "chats": [dict(r) for r in per_chat],
+            "db_size_bytes": os.path.getsize(self.db_path) if os.path.exists(self.db_path) else 0,
+        }
+
+    def chats(self) -> list[dict[str, Any]]:
+        with self._conn() as c:
+            rows = c.execute(
+                """SELECT chat_id, MAX(chat_title) AS title, MAX(chat_type) AS type,
+                          COUNT(*) AS n, MAX(timestamp_utc) AS last_ts
+                   FROM messages GROUP BY chat_id ORDER BY last_ts DESC"""
+            ).fetchall()
+        return [dict(r) for r in rows]
+
+    def _row_to_msg(self, row: sqlite3.Row) -> StoredMessage:
+        return StoredMessage(
+            id=row["id"],
+            chat_id=row["chat_id"],
+            chat_title=row["chat_title"],
+            message_id=row["message_id"],
+            user_id=row["user_id"],
+            user_display=self._user_display(row),
+            text=row["text"],
+            media_type=row["media_type"],
+            media_caption=row["media_caption"],
+            reply_to_message_id=row["reply_to_message_id"],
+            timestamp_utc=row["timestamp_utc"],
+            edit_timestamp_utc=row["edit_timestamp_utc"],
+        )
+
+
+def format_messages(msgs: Iterable[StoredMessage], tz_offset_hours: int = 0) -> str:
+    """Render messages as a chat-style transcript (chronological)."""
+    out: list[str] = []
+    for m in msgs:
+        ts = time.gmtime(m.timestamp_utc + tz_offset_hours * 3600)
+        stamp = time.strftime("%Y-%m-%d %H:%M", ts)
+        body = m.text or m.media_caption or (f"[{m.media_type}]" if m.media_type else "")
+        if m.media_type and (m.text or m.media_caption):
+            body = f"[{m.media_type}] {body}"
+        reply = f" ↩#{m.reply_to_message_id}" if m.reply_to_message_id else ""
+        edit = " (bearbeitet)" if m.edit_timestamp_utc else ""
+        out.append(f"[{stamp}] {m.user_display}{reply}{edit}: {body}")
+    return "\n".join(out)