diff --git a/chatlogger/README.md b/chatlogger/README.md new file mode 100644 index 000000000..b29d96030 --- /dev/null +++ b/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 ` | 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. diff --git a/chatlogger/__init__.py b/chatlogger/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/chatlogger/chatlogger-bot.service b/chatlogger/chatlogger-bot.service new file mode 100644 index 000000000..631cc9b74 --- /dev/null +++ b/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 diff --git a/chatlogger/chatlogger_bot.py b/chatlogger/chatlogger_bot.py new file mode 100644 index 000000000..87212dae8 --- /dev/null +++ b/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: {chat.id}\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"📒 Arakawa Concierge — Status", + f"DB: {store.db_path} ({s['db_size_bytes'] / 1024:.1f} KB)", + f"Nachrichten gesamt: {s['total_messages']}", + f"Edits: {s['total_edits']}", + "", + "Pro Chat:", + ] + 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" • {c['chat_id']} {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"
{_html_escape(chunk)}
", 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 <begriff>", 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"
{_html_escape(chunk)}
", 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("&", "&").replace("<", "<").replace(">", ">") + + +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() diff --git a/chatlogger/env.example b/chatlogger/env.example new file mode 100644 index 000000000..6be04ad93 --- /dev/null +++ b/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 diff --git a/chatlogger/query.py b/chatlogger/query.py new file mode 100644 index 000000000..2589a5158 --- /dev/null +++ b/chatlogger/query.py @@ -0,0 +1,143 @@ +"""CLI access to the chatlogger SQLite database. + +Usage: + chatlogger recent [--chat ] [-n 50] + chatlogger search "query string" [--chat ] [-n 30] + chatlogger stats + chatlogger chats + chatlogger since [--chat ] + chatlogger export [--chat ] [--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()) diff --git a/chatlogger/store.py b/chatlogger/store.py new file mode 100644 index 000000000..ede1ac5e2 --- /dev/null +++ b/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)