"""Hausmeister-Bot tool — read access to the chatlogger SQLite database. This file is INSTALLED into /opt/homelab-ai-bot/tools/chatlog.py (autodiscovered by tool_loader.py). It does NOT need a restart of the chatlogger bot — it only reads from the shared SQLite file (/opt/chatlogger/chats.db) using the Store class from the chatlogger package, which sits next to the hausmeister-bot in the same repo. Three LLM-callable functions: - chatlog_list_chats() → enumerate known group chats - chatlog_recent(limit, chat) → most recent N messages - chatlog_search(query, chat) → FTS5 search across messages - chatlog_since(minutes, chat) → time window """ from __future__ import annotations import os import sys import time from typing import Any # The chatlogger package sits next to homelab-ai-bot in the repo. # Both directories are bind-mounted into CT 116 (mp0 and mp1). _CHATLOGGER_PATH = "/opt/homelab-brain/chatlogger" if _CHATLOGGER_PATH not in sys.path: sys.path.insert(0, _CHATLOGGER_PATH) from store import Store, format_messages # noqa: E402 DB_PATH = os.environ.get("CHATLOGGER_DB", "/opt/chatlogger/chats.db") TZ_OFFSET_HOURS = int(os.environ.get("CHATLOGGER_TZ_OFFSET_HOURS", "7")) _store: Store | None = None def _get_store() -> Store: global _store if _store is None: _store = Store(DB_PATH) return _store def _resolve_chat(chat_filter: str | int | None) -> int | None: """Map a free-form chat hint (substring of title, or chat_id) to a real chat_id. Returns None when no filter or no match (callers treat None = 'all chats'). """ if chat_filter is None or chat_filter == "": return None try: return int(chat_filter) except (TypeError, ValueError): pass needle = str(chat_filter).lower().strip() for c in _get_store().chats(): title = (c.get("title") or "").lower() if needle in title: return c["chat_id"] return None # ---------------------------------------------------------------- Handlers def handle_chatlog_list_chats(**_kw) -> str: rows = _get_store().chats() if not rows: return "Keine Chats geloggt." lines = ["Bekannte Chats:"] for r in rows: last = time.strftime( "%Y-%m-%d %H:%M", time.gmtime(r["last_ts"] + TZ_OFFSET_HOURS * 3600), ) lines.append( f"- {r['title'] or '?'} (id={r['chat_id']}, {r['type']}, " f"{r['n']} Nachrichten, zuletzt {last})" ) return "\n".join(lines) def handle_chatlog_recent(limit: int = 30, chat: str | int | None = None, **_kw) -> str: limit = max(1, min(200, int(limit or 30))) chat_id = _resolve_chat(chat) msgs = _get_store().recent(chat_id=chat_id, limit=limit) if not msgs: if chat and chat_id is None: return f"Kein Chat gefunden, der zu '{chat}' passt." return "Keine Nachrichten gespeichert." header = _header(chat_id, len(msgs), label="letzte") return header + "\n" + format_messages(msgs, tz_offset_hours=TZ_OFFSET_HOURS) def handle_chatlog_search(query: str, chat: str | int | None = None, limit: int = 20, **_kw) -> str: if not query or not str(query).strip(): return "Suchbegriff fehlt." limit = max(1, min(200, int(limit or 20))) chat_id = _resolve_chat(chat) msgs = _get_store().search(query=str(query), chat_id=chat_id, limit=limit) if not msgs: return f"Keine Treffer für '{query}'." header = _header(chat_id, len(msgs), label=f"Treffer für '{query}'") return header + "\n" + format_messages(msgs, tz_offset_hours=TZ_OFFSET_HOURS) def handle_chatlog_since(minutes: int, chat: str | int | None = None, **_kw) -> str: minutes = max(1, min(60 * 24 * 30, int(minutes))) since_ts = int(time.time()) - minutes * 60 chat_id = _resolve_chat(chat) msgs = _get_store().since(chat_id=chat_id, since_ts=since_ts, limit=2000) if not msgs: return f"Keine Nachrichten in den letzten {minutes} Minuten." header = _header(chat_id, len(msgs), label=f"letzte {minutes} Minuten") return header + "\n" + format_messages(msgs, tz_offset_hours=TZ_OFFSET_HOURS) def _header(chat_id: int | None, n: int, label: str) -> str: if chat_id is None: return f"[{n} Nachrichten, {label}, alle Chats]" title = next( (c.get("title") for c in _get_store().chats() if c["chat_id"] == chat_id), None, ) return f"[{n} Nachrichten, {label}, Chat: {title or chat_id}]" # ---------------------------------------------------------------- LLM tool schema TOOLS = [ { "type": "function", "function": { "name": "chatlog_list_chats", "description": ( "Liste aller mitgeloggten Telegram-Gruppen-Chats mit Titel, " "ID, Anzahl Nachrichten und letztem Eintrag. Nutze dies, wenn " "unklar ist, welche Gruppe gemeint ist." ), "parameters": {"type": "object", "properties": {}, "required": []}, }, }, { "type": "function", "function": { "name": "chatlog_recent", "description": ( "Letzte N Nachrichten aus mitgeloggten Telegram-Gruppen-Chats " "(z.B. Airbnb-Concierge-Gruppe für die Wohnungen). Nutze dies " "bei Fragen wie 'Was ist der aktuelle Stand?', 'Was wurde " "zuletzt geschrieben?', 'Wer hat zuletzt was gesagt?', 'Muss " "ich auf etwas reagieren?'." ), "parameters": { "type": "object", "properties": { "limit": { "type": "integer", "description": "Anzahl letzter Nachrichten (default 30, max 200).", }, "chat": { "type": "string", "description": ( "Optional: Substring des Chat-Titels (z.B. 'Airbnb', " "'G 210', 'D1603') oder numerische chat_id. " "Leer = alle Chats." ), }, }, "required": [], }, }, }, { "type": "function", "function": { "name": "chatlog_search", "description": ( "Volltextsuche (FTS5) in den mitgeloggten Telegram-Chats. " "Nutze dies für Fragen wie 'Was wurde über X gesagt?', " "'Hat jemand X erwähnt?', 'Wer hat X zugesagt?'." ), "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": ( "Suchbegriff. FTS5-Syntax erlaubt: 'wort1 wort2' " "(beide), '\"phrase\"' (genau), 'wort*' (Prefix)." ), }, "chat": { "type": "string", "description": "Optional: Chat-Titel-Substring oder chat_id.", }, "limit": { "type": "integer", "description": "Max Treffer (default 20).", }, }, "required": ["query"], }, }, }, { "type": "function", "function": { "name": "chatlog_since", "description": ( "Alle Nachrichten der letzten N Minuten. Nutze dies bei " "Fragen wie 'Was war heute los?', 'Was war in der letzten " "Stunde?'." ), "parameters": { "type": "object", "properties": { "minutes": { "type": "integer", "description": "Zeitfenster in Minuten (z.B. 60 = letzte Stunde, 1440 = letzte 24h).", }, "chat": { "type": "string", "description": "Optional: Chat-Titel-Substring oder chat_id.", }, }, "required": ["minutes"], }, }, }, ] HANDLERS = { "chatlog_list_chats": handle_chatlog_list_chats, "chatlog_recent": handle_chatlog_recent, "chatlog_search": handle_chatlog_search, "chatlog_since": handle_chatlog_since, } SYSTEM_PROMPT_EXTRA = """ Telegram-Gruppen-Mitschnitt: Es gibt einen passiven Mitleser-Bot (@arakawa_concierge_bot), der bestimmte Telegram-Gruppen aufzeichnet — z.B. die Airbnb-Concierge-Gruppe für die ARAKAWA-Wohnungen (G2010B, D1603). Wenn der Benutzer fragt nach "aktueller Stand", "wer hat was zugesagt", "was ist offen", "muss ich reagieren", "letzte Nachricht im Chat", "Concierge", "Airbnb", "Listing", "Vermietung" — verwende die chatlog_*-Tools. Bei Unklarheit über die gemeinte Gruppe zuerst chatlog_list_chats() aufrufen. """.strip()